### Question:
RuntimeClientService extends AbstractRetryableScheduledService { @VisibleForTesting long getProgramFinishTime() { return programFinishTime; } @Inject  RuntimeClientService(CConfiguration cConf, MessagingService messagingService,
                       RuntimeClient runtimeClient, ProgramRunId programRunId);   }### Answer:
@Test public void testProgramTerminate() throws Exception { MessagingContext messagingContext = new MultiThreadMessagingContext(clientMessagingService); MessagePublisher messagePublisher = messagingContext.getDirectMessagePublisher(); ProgramStateWriter programStateWriter = new MessagingProgramStateWriter(clientCConf, clientMessagingService); programStateWriter.completed(PROGRAM_RUN_ID); Tasks.waitFor(true, () -> runtimeClientService.getProgramFinishTime() >= 0, 2, TimeUnit.SECONDS); for (Map.Entry<String, String> entry : topicConfigs.entrySet()) { if (!entry.getKey().equals(Constants.AppFabric.PROGRAM_STATUS_EVENT_TOPIC)) { List<String> payloads = Arrays.asList(entry.getKey(), entry.getKey(), entry.getKey()); messagePublisher.publish(NamespaceId.SYSTEM.getNamespace(), entry.getValue(), StandardCharsets.UTF_8, payloads.iterator()); } } Tasks.waitFor(Service.State.TERMINATED, () -> runtimeClientService.state(), clientCConf.getLong(Constants.RuntimeMonitor.GRACEFUL_SHUTDOWN_MS) + 2000, TimeUnit.MILLISECONDS); MessagingContext serverMessagingContext = new MultiThreadMessagingContext(messagingService); for (Map.Entry<String, String> entry : topicConfigs.entrySet()) { if (entry.getKey().equals(Constants.AppFabric.PROGRAM_STATUS_EVENT_TOPIC)) { Tasks.waitFor(Collections.singletonList(ProgramRunStatus.COMPLETED), () -> fetchMessages(serverMessagingContext, entry.getValue(), 10, null).stream() .map(Message::getPayloadAsString) .map(s -> GSON.fromJson(s, Notification.class)) .map(n -> n.getProperties().get(ProgramOptionConstants.PROGRAM_STATUS)) .map(ProgramRunStatus::valueOf) .collect(Collectors.toList()), 5, TimeUnit.SECONDS); } else { Tasks.waitFor(Arrays.asList(entry.getKey(), entry.getKey(), entry.getKey()), () -> fetchMessages(serverMessagingContext, entry.getValue(), 10, null) .stream().map(Message::getPayloadAsString).collect(Collectors.toList()), 5, TimeUnit.SECONDS); } } }