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();
    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();
        }
      }

      // Add all checkpointed residuals to the response.
      response.addAllResidualRoots(bundleProcessor.getSplitListener().getResidualRoots());

      // TODO(BEAM-6597): This should be reporting monitoring infos using the short id system.
      // Get start bundle Execution Time Metrics.
      response.addAllMonitoringInfos(
          bundleProcessor.getStartFunctionRegistry().getExecutionTimeMonitoringInfos());
      // Get process bundle Execution Time Metrics.
      response.addAllMonitoringInfos(
          bundleProcessor.getpCollectionConsumerRegistry().getExecutionTimeMonitoringInfos());
      // Get finish bundle Execution Time Metrics.
      response.addAllMonitoringInfos(
          bundleProcessor.getFinishFunctionRegistry().getExecutionTimeMonitoringInfos());
      // Extract MonitoringInfos that come from the metrics container registry.
      response.addAllMonitoringInfos(
          bundleProcessor.getMetricsContainerRegistry().getMonitoringInfos());
      // Add any additional monitoring infos that the "runners" report explicitly.
      for (ProgressRequestCallback progressRequestCallback :
          bundleProcessor.getProgressRequestCallbacks()) {
        response.addAllMonitoringInfos(progressRequestCallback.getMonitoringInfos());
      }

      if (!bundleProcessor.getBundleFinalizationCallbackRegistrations().isEmpty()) {
        finalizeBundleHandler.registerCallbacks(
            bundleProcessor.getInstructionId(),
            ImmutableList.copyOf(bundleProcessor.getBundleFinalizationCallbackRegistrations()));
        response.setRequiresFinalization(true);
      }

      bundleProcessorCache.release(
          request.getProcessBundle().getProcessBundleDescriptorId(), bundleProcessor);
    }
    return BeamFnApi.InstructionResponse.newBuilder().setProcessBundle(response);
  }

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

}