RegisterAndProcessBundleOperation extends Operation { public CompletionStage<BeamFnApi.ProcessBundleProgressResponse> getProcessBundleProgress() { String processBundleId = this.processBundleId; if (processBundleId == null) { return CompletableFuture.completedFuture( BeamFnApi.ProcessBundleProgressResponse.getDefaultInstance()); } InstructionRequest processBundleRequest = InstructionRequest.newBuilder() .setInstructionId(idGenerator.getId()) .setProcessBundleProgress( ProcessBundleProgressRequest.newBuilder().setInstructionId(processBundleId)) .build(); return instructionRequestHandler .handle(processBundleRequest) .thenApply(InstructionResponse::getProcessBundleProgress); } RegisterAndProcessBundleOperation(
      IdGenerator idGenerator,
      InstructionRequestHandler instructionRequestHandler,
      StateDelegator beamFnStateDelegator,
      RegisterRequest registerRequest,
      Map<String, DataflowOperationContext> ptransformIdToOperationContext,
      Map<String, DataflowStepContext> ptransformIdToSystemStepContext,
      Map<String, SideInputReader> ptransformIdToSideInputReader,
      Table<String, String, PCollectionView<?>> ptransformIdToSideInputIdToPCollectionView,
      Map<String, NameContext> pcollectionIdToNameContext,
      OperationContext context); synchronized String getProcessBundleInstructionId(); String getCurrentProcessBundleInstructionId(); @Override void start(); @Override void finish(); @Override void abort(); Map<String, NameContext> getPCollectionIdToNameContext(); Map<String, DataflowStepContext> getPtransformIdToUserStepContext(); CompletionStage<BeamFnApi.ProcessBundleProgressResponse> getProcessBundleProgress(); CompletionStage<List<MonitoringInfo>> getFinalMonitoringInfos(); boolean hasFailed(); List<MonitoringInfo> findIOPCollectionMonitoringInfos(
      Iterable<MonitoringInfo> monitoringInfos); @Override boolean supportsRestart(); @Override String toString();  }
@Test public void testGetProcessBundleProgressReturnsDefaultInstanceIfNoBundleIdCached() throws Exception { InstructionRequestHandler mockInstructionRequestHandler = mock(InstructionRequestHandler.class); RegisterAndProcessBundleOperation operation = new RegisterAndProcessBundleOperation( IdGenerators.decrementingLongs(), mockInstructionRequestHandler, mockBeamFnStateDelegator, REGISTER_REQUEST, ImmutableMap.of(), ImmutableMap.of(), ImmutableMap.of(), ImmutableTable.of(), ImmutableMap.of(), mockContext); assertEquals( ProcessBundleProgressResponse.getDefaultInstance(), MoreFutures.get(operation.getProcessBundleProgress())); }