@VisibleForTesting
  String getActiveProcessBundleState() {
    StringJoiner activeBundlesState = new StringJoiner("\n");
    activeBundlesState.add("========== ACTIVE PROCESSING BUNDLES ==========");
    if (processBundleCache.getActiveBundleProcessors().isEmpty()) {
      activeBundlesState.add("No active processing bundles.");
    } else {
      List<BundleState> bundleStates = new ArrayList<>();
      processBundleCache.getActiveBundleProcessors().entrySet().stream()
          .forEach(
              instructionAndBundleProcessor -> {
                BundleProcessor bundleProcessor = instructionAndBundleProcessor.getValue();
                ExecutionStateTrackerStatus executionStateTrackerStatus =
                    bundleProcessor.getStateTracker().getStatus();
                if (executionStateTrackerStatus != null) {
                  bundleStates.add(
                      new BundleState(
                          instructionAndBundleProcessor.getKey(),
                          executionStateTrackerStatus.getTrackedThread().getName(),
                          DateTimeUtils.currentTimeMillis()
                              - executionStateTrackerStatus.getLastTransitionTimeMillis()));
                }
              });
      bundleStates.stream()
          // reverse sort active bundle by time since last transition.
          .sorted(Comparator.comparing(BundleState::getTimeSinceTransition).reversed())
          .limit(10) // only keep top 10
          .forEachOrdered(
              bundleState -> {
                activeBundlesState.add(
                    String.format("---- Instruction %s ----", bundleState.getInstruction()));
                activeBundlesState.add(
                    String.format("Tracked thread: %s", bundleState.getTrackedThreadName()));
                activeBundlesState.add(
                    String.format(
                        "Time since transition: %.2f seconds%n",
                        bundleState.getTimeSinceTransition() / 1000.0));
              });
    }
    return activeBundlesState.toString();
  }
@Test
  public void testActiveBundleState() {
    ProcessBundleHandler handler = mock(ProcessBundleHandler.class);
    BundleProcessorCache processorCache = mock(BundleProcessorCache.class);
    Map<String, BundleProcessor> bundleProcessorMap = new HashMap<>();
    for (int i = 0; i < 11; i++) {
      BundleProcessor processor = mock(BundleProcessor.class);
      ExecutionStateTracker executionStateTracker = mock(ExecutionStateTracker.class);
      when(processor.getStateTracker()).thenReturn(executionStateTracker);
      when(executionStateTracker.getStatus())
          .thenReturn(
              ExecutionStateTrackerStatus.create(
                  "ptransformId", "ptransformIdName", Thread.currentThread(), i * 1000, null));
      String instruction = Integer.toString(i);
      when(processorCache.find(instruction)).thenReturn(processor);
      bundleProcessorMap.put(instruction, processor);
    }
    when(handler.getBundleProcessorCache()).thenReturn(processorCache);
    when(processorCache.getActiveBundleProcessors()).thenReturn(bundleProcessorMap);

    ManagedChannelFactory channelFactory = ManagedChannelFactory.createInProcess();
    BeamFnStatusClient client =
        new BeamFnStatusClient(
            apiServiceDescriptor,
            channelFactory::forDescriptor,
            handler.getBundleProcessorCache(),
            PipelineOptionsFactory.create(),
            Caches.noop());
    StringJoiner joiner = new StringJoiner("\n");
    joiner.add(client.getActiveProcessBundleState());
    String actualState = joiner.toString();

    List<String> expectedInstructions = new ArrayList<>();
    for (int i = 0; i < 10; i++) {
      expectedInstructions.add(String.format("Instruction %d", i));
    }
    assertThat(actualState, stringContainsInOrder(expectedInstructions));
    assertThat(actualState, not(containsString("Instruction 10")));
  }