KafkaConsumerConfig extends HashMap<String, Object> { public static String createClientId(String prefix, Config config) { Pair<String, String> jobNameId = getJobNameAndId(config); String jobName = jobNameId.getLeft(); String jobId = jobNameId.getRight(); return String.format("%s-%s-%s", prefix.replaceAll("\\W", "_"), jobName.replaceAll("\\W", "_"), jobId.replaceAll("\\W", "_")); } private  KafkaConsumerConfig(Map<String, Object> props, String systemName); static KafkaConsumerConfig getKafkaSystemConsumerConfig(Config config, String systemName, String clientId); String getClientId(); int fetchMessageMaxBytes(); String getZkConnect(); static String createClientId(String prefix, Config config); static Pair<String, String> getJobNameAndId(Config config); static final Logger LOG; static final String ZOOKEEPER_CONNECT; }
@Test public void testGetConsumerClientId() { Map<String, String> map = new HashMap<>(); map.put(JobConfig.JOB_NAME, "jobName"); map.put(JobConfig.JOB_ID, "jobId"); String result = KafkaConsumerConfig.createClientId("consumer", new MapConfig(map)); Assert.assertEquals("consumer-jobName-jobId", result); result = KafkaConsumerConfig.createClientId("consumer-", new MapConfig(map)); Assert.assertEquals("consumer_-jobName-jobId", result); result = KafkaConsumerConfig.createClientId("super-duper-consumer", new MapConfig(map)); Assert.assertEquals("super_duper_consumer-jobName-jobId", result); map.put(JobConfig.JOB_NAME, " very important!job"); result = KafkaConsumerConfig.createClientId("consumer", new MapConfig(map)); Assert.assertEquals("consumer-_very_important_job-jobId", result); map.put(JobConfig.JOB_ID, "number-#3"); result = KafkaConsumerConfig.createClientId("consumer", new MapConfig(map)); Assert.assertEquals("consumer-_very_important_job-number__3", result); }