Summary & Motivation

Allow run time configuration of dynamic steps using the k8s_job_executor. Previously, dynamic steps would not be configured when using per_step_k8s_config. Now doing something along line of the following:

execution:
  config:
    step_k8s_config:
      dyn_sink:
         container_config:
           resources:
             requests:
               cpu: '300m'

Given the following job:

@job
def dynamic_producer_consumer_job():
    @op(out=DynamicOut(int))
    def dyn_producer():
    for i in range(3):
        return DynamicOutput(i, str(i))
    @op
    def dyn_sink(producer: int): ...

Will configure all dyn_sink ops with the following modified config.

cc @Kuhlwein

resolves #26588

How I Tested These Changes

Red green coverage given a dynamic job. Revert the following line to watch the test fail.