title: Airflow 2.0.1: Pod Template Override not working as expected for KubernetesExecutor
body: <p>Setup: Airflow 2.0.1 with Kubernetes 1.18 and Python 3.8, Kubernetes Client: 18.17.x</p>
  <p>Pod template file:</p>
  <pre><code>apiVersion: v1
  kind: Pod
  metadata:
    name: workerPod
  
  spec:
    containers:
      - args: []
        command: []
        env:
          - name: &lt;Key&gt;
            value: &quot;&lt;value&gt;&quot;
        envFrom: []
        name: base
        image: &quot;&lt;image_name&gt;&quot;
        imagePullSecrets: [name: &quot;&lt;image_pull_secrets&gt;&quot;]
        imagePullPolicy: &quot;Always&quot;
        ports: []
        volumeMounts:
          - mountPath: &quot;&lt;path&gt;&quot;
            name: &quot;&lt;name&gt;&quot;
  </code></pre>
  <p>The default config set in airflow.cfg is as follows:</p>
  <pre><code>[kubernetes]
  pod_template_file = &lt;path to template file&gt;
  worker_container_repository = &lt;base-default-image&gt;
  worker_container_tag = &lt;tag&gt;
  namespace = airflow
  delete_worker_pods = True
  delete_worker_pods_on_failure = False
  worker_pods_creation_batch_size = 1
  multi_namespace_mode = False
  in_cluster = True
  kube_client_request_args =
  delete_option_kwargs =
  enable_tcp_keepalive = False
  tcp_keep_idle = 120
  tcp_keep_intvl = 30
  tcp_keep_cnt = 6
  dags_in_image = True
  dags_volume_mount_point = &lt;volume-mount-point&gt;
  image_pull_secrets = &lt;default-pull-secrets&gt;
  </code></pre>
  <p>The problem is that, while certain keys are being read correctly from the pod_template_file, for instance, I can see all the <code>env variables</code> be set correctly as well as <code>imagePullPolicy</code> being read correctly as well (validated by overriding value of <code>imagePullPolicy: &quot;Always&quot;</code> from <code>imagePullPolicy: &quot;IfNotPresent&quot;</code>), but the key for <code>imagePullSecrets</code> is not being read correctly. I can validate this, as I get a <code>Base credentials not provided</code> error when the image is being pulled from the ecr repo. I have validated that the credentials are correct and I can create a pod when trying to do so explicitly.</p>
  <p>Even when trying to set the <code>imagePullSecrets</code> in the <code>airflow.cfg</code> directly, I still end up getting the same error.</p>
  <p>I have also tried to create the pod override using the V1 api explicitly as follows:</p>
  <pre><code>start_task = PythonOperator(
              task_id=&lt;start_task_id&gt;, python_callable=&lt;start_task_callabel&gt;, op_args=[&lt;args&gt;], dag=dag,
              executor_config={
                  &quot;pod_template_file&quot;: &quot;&lt;path_to_template&gt;&quot;,
                  &quot;pod_override&quot;: k8s.V1Pod(
                      spec=k8s.V1PodSpec(
                          containers=[
                              k8s.V1Container(
                                  name=&quot;base&quot;,
                                  image=&quot;&lt;image_override&gt;&quot;,
                                  image_pull_policy=&quot;&lt;pull_policy&gt;&quot;
                              ),
                          ],
                          image_pull_secrets=[k8s.V1LocalObjectReference('&lt;image_pull_secrets&gt;')],
                      )
                  ),
              },
          )
  </code></pre>
  <p>In this case I can get the docker image to be used downloaded correctly without any authentication errors. But unfortunately, the pod throws an error: <code>AttributeError: 'V1Container' object has no attribute '_startup_probe'</code></p>
  <pre><code>Traceback (most recent call last):
    File &quot;/usr/local/bin/airflow&quot;, line 8, in &lt;module&gt;
      sys.exit(main())
    File &quot;/usr/local/lib/python3.8/dist-packages/airflow/__main__.py&quot;, line 40, in main
      args.func(args)
    File &quot;/usr/local/lib/python3.8/dist-packages/airflow/cli/cli_parser.py&quot;, line 48, in command
      return func(*args, **kwargs)
    File &quot;/usr/local/lib/python3.8/dist-packages/airflow/utils/cli.py&quot;, line 89, in wrapper
      return f(*args, **kwargs)
    File &quot;/usr/local/lib/python3.8/dist-packages/airflow/cli/commands/task_command.py&quot;, line 234, in task_run
      _run_task_by_selected_method(args, dag, ti)
    File &quot;/usr/local/lib/python3.8/dist-packages/airflow/cli/commands/task_command.py&quot;, line 64, in _run_task_by_selected_method
      _run_task_by_local_task_job(args, ti)
    File &quot;/usr/local/lib/python3.8/dist-packages/airflow/cli/commands/task_command.py&quot;, line 120, in _run_task_by_local_task_job
      run_job.run()
    File &quot;/usr/local/lib/python3.8/dist-packages/airflow/jobs/base_job.py&quot;, line 237, in run
      self._execute()
    File &quot;/usr/local/lib/python3.8/dist-packages/airflow/jobs/local_task_job.py&quot;, line 84, in _execute
      if not self.task_instance.check_and_change_state_before_execution(
    File &quot;/usr/local/lib/python3.8/dist-packages/airflow/utils/session.py&quot;, line 65, in wrapper
      return func(*args, session=session, **kwargs)
    File &quot;/usr/local/lib/python3.8/dist-packages/airflow/models/taskinstance.py&quot;, line 1029, in check_and_change_state_before_execution
      session.commit()
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/orm/session.py&quot;, line 1046, in commit
      self.transaction.commit()
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/orm/session.py&quot;, line 504, in commit
      self._prepare_impl()
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/orm/session.py&quot;, line 483, in _prepare_impl
      self.session.flush()
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/orm/session.py&quot;, line 2540, in flush
      self._flush(objects)
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/orm/session.py&quot;, line 2682, in _flush
      transaction.rollback(_capture_exception=True)
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/util/langhelpers.py&quot;, line 68, in __exit__
      compat.raise_(
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/util/compat.py&quot;, line 182, in raise_
      raise exception
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/orm/session.py&quot;, line 2642, in _flush
      flush_context.execute()
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/orm/unitofwork.py&quot;, line 422, in execute
      rec.execute(self)
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/orm/unitofwork.py&quot;, line 586, in execute
      persistence.save_obj(
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/orm/persistence.py&quot;, line 230, in save_obj
      _emit_update_statements(
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/orm/persistence.py&quot;, line 885, in _emit_update_statements
      for (
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/orm/persistence.py&quot;, line 626, in _collect_update_commands
      state.manager[propkey].impl.is_equal(
    File &quot;/usr/local/lib/python3.8/dist-packages/sqlalchemy/sql/sqltypes.py&quot;, line 1738, in compare_values
      return x == y
    File &quot;/usr/local/lib/python3.8/dist-packages/kubernetes/client/models/v1_pod.py&quot;, line 221, in __eq__
      return self.to_dict() == other.to_dict()
    File &quot;/usr/local/lib/python3.8/dist-packages/kubernetes/client/models/v1_pod.py&quot;, line 196, in to_dict
      result[attr] = value.to_dict()
    File &quot;/usr/local/lib/python3.8/dist-packages/kubernetes/client/models/v1_pod_spec.py&quot;, line 1004, in to_dict
      result[attr] = list(map(
    File &quot;/usr/local/lib/python3.8/dist-packages/kubernetes/client/models/v1_pod_spec.py&quot;, line 1005, in &lt;lambda&gt;
      lambda x: x.to_dict() if hasattr(x, &quot;to_dict&quot;) else x,
    File &quot;/usr/local/lib/python3.8/dist-packages/kubernetes/client/models/v1_container.py&quot;, line 660, in to_dict
      value = getattr(self, attr)
    File &quot;/usr/local/lib/python3.8/dist-packages/kubernetes/client/models/v1_container.py&quot;, line 458, in startup_probe
      return self._startup_probe
  AttributeError: 'V1Container' object has no attribute '_startup_probe'
  </code></pre>