I am working on a microservice architecture using Spring boot. One of my application act as a Kafka producer to write logs on a Kafka topic. My producer config is as follow:
@Configuration
public class KafkaProducerConfig {
@Value("${kafka.address:X.X.X.X:9092}")
private String bootstrapAddress;
@Bean
public ProducerFactory<String, KafkaTopicMessage> producerFactory() {
Map<String, Object> configProps = new HashMap<>();
configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringOrBytesSerializer.class);
configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
return new DefaultKafkaProducerFactory<>(configProps);
}
@Bean
public KafkaTemplate<String, KafkaTopicMessage> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}
Also, my topic config is :
@Configuration
public class KafkaTopicConfig {
@Value("${kafka.address:X.X.X.X:9092}")
private String bootstrapAddress;
@Bean
public KafkaAdmin kafkaAdmin() {
Map<String, Object> configs = new HashMap<>();
configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
return new KafkaAdmin(configs);
}
@Bean
public NewTopic logTopic() {
return new NewTopic("log-topic", 1, (short) 1);
}
}
And Another application is consuming this topic. My consumer Config is like below:
@EnableKafka
@Configuration
public class KafkaConsumerConfig {
@Bean
public ConcurrentKafkaListenerContainerFactory<String, KafkaTopicMessage> concurrentKafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, KafkaTopicMessage> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
@Bean
public ConsumerFactory<String, KafkaTopicMessage> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "X.X.X.X:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "log");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
return new DefaultKafkaConsumerFactory<>(props);
}
}
I have also included the configuration application.yml file. like this:
kafka:
address: X.X.X.X:9092
spring:
kafka:
consumer:
bootstrap-servers: ${kafka.address}
group-id: log
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
properties:
spring:
json:
trusted:
packages: '*'
producer:
bootstrap-servers: ${kafka.address}
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
The problem is that even though I'm giving my server addresses explicitly it is trying to connect to 127.0.0.1:9092. Can anybody help me with this problem? why it is not reading my config and why it's going to have a default config?
I appreciate your help.