KafkaUtil { public static String getCheckpointTopic(String jobName, String jobId, Config config) { String checkpointTopic = String.format(CHECKPOINT_TOPIC_FORMAT, CHECKPOINT_LOG_VERSION_NUMBER, jobName.replaceAll("_", "-"), jobId.replaceAll("_", "-")); return StreamManager.createUniqueNameForBatch(checkpointTopic, config); }  static String getCheckpointTopic(String jobName, String jobId, Config config); static Integer getIntegerPartitionKey(OutgoingMessageEnvelope envelope, List<PartitionInfo> partitions); static SystemStreamPartition toSystemStreamPartition(String systemName, TopicPartition topicPartition);  }
@Test public void testGetCheckpointTopic() { assertEquals("__samza_checkpoint_ver_1_for_myJob_myId", KafkaUtil.getCheckpointTopic("myJob", "myId", new MapConfig())); assertEquals("__samza_checkpoint_ver_1_for_my-job-with-underscore_my-id-with-underscore", KafkaUtil.getCheckpointTopic("my_job-with_underscore", "my_id-with_underscore", new MapConfig())); Config configWithBatchMode = new MapConfig(ImmutableMap.of( ApplicationConfig.APP_MODE, ApplicationConfig.ApplicationMode.BATCH.name(), ApplicationConfig.APP_RUN_ID, "myRunId" )); assertEquals("__samza_checkpoint_ver_1_for_myJob_myId-myRunId", KafkaUtil.getCheckpointTopic("myJob", "myId", configWithBatchMode)); assertEquals("__samza_checkpoint_ver_1_for_my-job-with-underscore_my-id-with-underscore-myRunId", KafkaUtil.getCheckpointTopic("my_job-with_underscore", "my_id-with_underscore", configWithBatchMode)); }