ProcessBundleHandler { public BeamFnApi.InstructionResponse.Builder trySplit(BeamFnApi.InstructionRequest request) { BundleProcessor bundleProcessor = bundleProcessorCache.find(request.getProcessBundleSplit().getInstructionId()); BeamFnApi.ProcessBundleSplitResponse.Builder response = BeamFnApi.ProcessBundleSplitResponse.newBuilder(); if (bundleProcessor == null) { return BeamFnApi.InstructionResponse.newBuilder() .setProcessBundleSplit(BeamFnApi.ProcessBundleSplitResponse.getDefaultInstance()); } for (BeamFnDataReadRunner channelRoot : bundleProcessor.getChannelRoots()) { channelRoot.trySplit(request.getProcessBundleSplit(), response); } return BeamFnApi.InstructionResponse.newBuilder().setProcessBundleSplit(response); } 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); BeamFnApi.InstructionResponse.Builder processBundle(BeamFnApi.InstructionRequest request); BeamFnApi.InstructionResponse.Builder progress(BeamFnApi.InstructionRequest request); BeamFnApi.InstructionResponse.Builder trySplit(BeamFnApi.InstructionRequest request); void shutdown(); static final String JAVA_SOURCE_URN; }
@Test public void testTrySplitBeforeBundleDoesNotFail() { ProcessBundleHandler handler = new ProcessBundleHandler( PipelineOptionsFactory.create(), null, beamFnDataClient, null , null , ImmutableMap.of(), new BundleProcessorCache()); BeamFnApi.InstructionResponse response = handler .trySplit( BeamFnApi.InstructionRequest.newBuilder() .setInstructionId("999L") .setProcessBundleSplit( BeamFnApi.ProcessBundleSplitRequest.newBuilder() .setInstructionId("unknown-id")) .build()) .build(); assertNotNull(response.getProcessBundleSplit()); assertEquals(0, response.getProcessBundleSplit().getChannelSplitsCount()); }