@Override
  public void run() {
    MetricsContainerImpl metricsContainer = new MetricsContainerImpl(transform.getFullName());
    try (Closeable metricsScope = MetricsEnvironment.scopedMetricsContainer(metricsContainer)) {
      Collection<ModelEnforcement<T>> enforcements = new ArrayList<>();
      for (ModelEnforcementFactory enforcementFactory : modelEnforcements) {
        ModelEnforcement<T> enforcement = enforcementFactory.forBundle(inputBundle, transform);
        enforcements.add(enforcement);
      }
      @Nullable
      TransformEvaluator<T> evaluator = evaluatorRegistry.forApplication(transform, inputBundle);
      if (evaluator == null) {
        onComplete.handleEmpty(transform);
        // Nothing to do
        return;
      }

      processElements(evaluator, metricsContainer, enforcements);

      finishBundle(evaluator, metricsContainer, enforcements);
    } catch (Exception e) {
      onComplete.handleException(inputBundle, e);
      if (e instanceof RuntimeException) {
        throw (RuntimeException) e;
      }
      throw new RuntimeException(e);
    } catch (Error err) {
      LOG.error("Error occurred within {}", this, err);
      onComplete.handleError(err);
      throw err;
    } finally {
      // Report the physical metrics from the end of this step.
      context.getMetrics().commitPhysical(inputBundle, metricsContainer.getCumulative());

      transformEvaluationState.complete(this);
    }
  }
@Test
  public void callWithNullInputBundleFinishesBundleAndCompletes() throws Exception {
    final TransformResult<Object> result = StepTransformResult.withoutHold(createdProducer).build();
    final AtomicBoolean finishCalled = new AtomicBoolean(false);
    TransformEvaluator<Object> evaluator =
        new TransformEvaluator<Object>() {
          @Override
          public void processElement(WindowedValue<Object> element) throws Exception {
            throw new IllegalArgumentException("Shouldn't be called");
          }

          @Override
          public TransformResult<Object> finishBundle() throws Exception {
            finishCalled.set(true);
            return result;
          }
        };

    when(registry.forApplication(createdProducer, null)).thenReturn(evaluator);

    DirectTransformExecutor<Object> executor =
        new DirectTransformExecutor<>(
            evaluationContext,
            registry,
            Collections.emptyList(),
            null,
            createdProducer,
            completionCallback,
            transformEvaluationState);
    executor.run();

    assertThat(finishCalled.get(), is(true));
    assertThat(completionCallback.handledResult, equalTo(result));
    assertThat(completionCallback.handledException, is(nullValue()));
  }