class ProcessBundleHandler {

  public BeamFnApi.InstructionResponse.Builder processBundle(BeamFnApi.InstructionRequest request)
      throws Exception {
    BeamFnApi.ProcessBundleResponse.Builder response = BeamFnApi.ProcessBundleResponse.newBuilder();

    BundleProcessor bundleProcessor =
        bundleProcessorCache.get(
            request.getProcessBundle().getProcessBundleDescriptorId(),
            request.getInstructionId(),
            () -> {
              try {
                return createBundleProcessor(
                    request.getProcessBundle().getProcessBundleDescriptorId(),
                    request.getProcessBundle());
              } catch (IOException e) {
                throw new RuntimeException(e);
              }
            });
    PTransformFunctionRegistry startFunctionRegistry = bundleProcessor.getStartFunctionRegistry();
    PTransformFunctionRegistry finishFunctionRegistry = bundleProcessor.getFinishFunctionRegistry();
    Multimap<String, DelayedBundleApplication> allResiduals = bundleProcessor.getAllResiduals();
    PCollectionConsumerRegistry pCollectionConsumerRegistry =
        bundleProcessor.getpCollectionConsumerRegistry();
    MetricsContainerStepMap metricsContainerRegistry =
        bundleProcessor.getMetricsContainerRegistry();
    ExecutionStateTracker stateTracker = bundleProcessor.getStateTracker();
    QueueingBeamFnDataClient queueingClient = bundleProcessor.getQueueingClient();

    try (HandleStateCallsForBundle beamFnStateClient = bundleProcessor.getBeamFnStateClient()) {
      try (Closeable closeTracker = stateTracker.activate()) {
        // Already in reverse topological order so we don't need to do anything.
        for (ThrowingRunnable startFunction : startFunctionRegistry.getFunctions()) {
          LOG.debug("Starting function {}", startFunction);
          startFunction.run();
        }

        queueingClient.drainAndBlock();

        // Need to reverse this since we want to call finish in topological order.
        for (ThrowingRunnable finishFunction :
            Lists.reverse(finishFunctionRegistry.getFunctions())) {
          LOG.debug("Finishing function {}", finishFunction);
          finishFunction.run();
        }
        if (!allResiduals.isEmpty()) {
          response.addAllResidualRoots(allResiduals.values());
        }
      }
      // Get start bundle Execution Time Metrics.
      for (MonitoringInfo mi : startFunctionRegistry.getExecutionTimeMonitoringInfos()) {
        response.addMonitoringInfos(mi);
      }
      // Get process bundle Execution Time Metrics.
      for (MonitoringInfo mi : pCollectionConsumerRegistry.getExecutionTimeMonitoringInfos()) {
        response.addMonitoringInfos(mi);
      }

      // Get finish bundle Execution Time Metrics.
      for (MonitoringInfo mi : finishFunctionRegistry.getExecutionTimeMonitoringInfos()) {
        response.addMonitoringInfos(mi);
      }
      // Extract all other MonitoringInfos other than the execution time monitoring infos.
      for (MonitoringInfo mi : metricsContainerRegistry.getMonitoringInfos()) {
        response.addMonitoringInfos(mi);
      }
      bundleProcessorCache.release(
          request.getProcessBundle().getProcessBundleDescriptorId(), bundleProcessor);
    }
    return BeamFnApi.InstructionResponse.newBuilder().setProcessBundle(response);
  }

  public  ProcessBundleHandler(
      PipelineOptions options,
      Function<String, Message> fnApiRegistry,
      BeamFnDataClient beamFnDataClient,
      BeamFnStateGrpcClientCache beamFnStateGrpcClientCache);
  @VisibleForTesting  ProcessBundleHandler(
      PipelineOptions options,
      Function<String, Message> fnApiRegistry,
      BeamFnDataClient beamFnDataClient,
      BeamFnStateGrpcClientCache beamFnStateGrpcClientCache,
      Map<String, PTransformRunnerFactory> urnToPTransformRunnerFactoryMap,
      BundleProcessorCache bundleProcessorCache);

}