KafkaConfig { @Bean public Map<String, Object> consumerProperties() { final Map<String, Object> props = new HashMap<>(); props.put("bootstrap.servers", environment.getProperty(MetronRestConstants.KAFKA_BROKER_URL_SPRING_PROPERTY)); props.put("group.id", "metron-rest"); props.put("enable.auto.commit", "false"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "30000"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); if (environment.getProperty(MetronRestConstants.KERBEROS_ENABLED_SPRING_PROPERTY, Boolean.class, false)) { props.put("security.protocol", KafkaUtils.INSTANCE.normalizeProtocol(environment.getProperty(MetronRestConstants.KAFKA_SECURITY_PROTOCOL_SPRING_PROPERTY))); } return props; } @Autowired  KafkaConfig(final Environment environment); @Bean ZkUtils zkUtils(); @Bean Map<String, Object> consumerProperties(); @Bean ConsumerFactory<String, String> createConsumerFactory(); @Bean Map<String, Object> producerProperties(); @Bean KafkaProducer kafkaProducer(); @Bean AdminUtils adminUtils();  }
@Test public void kafkaConfigShouldProperlyReturnConsumerProperties() { when(environment.getProperty(MetronRestConstants.KAFKA_BROKER_URL_SPRING_PROPERTY)).thenReturn("broker urls"); when(environment.getProperty(MetronRestConstants.KERBEROS_ENABLED_SPRING_PROPERTY, Boolean.class, false)).thenReturn(false); Map<String, Object> consumerProperties = kafkaConfig.consumerProperties(); assertEquals("broker urls", consumerProperties.get("bootstrap.servers")); assertNull(consumerProperties.get("security.protocol")); when(environment.getProperty(MetronRestConstants.KERBEROS_ENABLED_SPRING_PROPERTY, Boolean.class, false)).thenReturn(true); when(environment.getProperty(MetronRestConstants.KAFKA_SECURITY_PROTOCOL_SPRING_PROPERTY)).thenReturn("kafka security protocol"); consumerProperties = kafkaConfig.consumerProperties(); assertEquals("kafka security protocol", consumerProperties.get("security.protocol")); }