A simple spring-boot-kafka which consumes from a topic on a network cluster:
Errors:
Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected
Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
Puzzle:
The configured broker is not local, it's BROKER_1.FOO.NET:9094, and it is available.
pom.xml
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.7.1</version>
</dependency>
config class:
@Slf4j
@EnableKafka
@Configuration
@PropertySource("dv/application.properties")
public class KafkaConfig {
@Bean
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> config = new HashMap<>();
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "BROKER_1.FOO.NET:9094");
config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_1");
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
return new DefaultKafkaConsumerFactory<>(config);
}
@Bean
public ConcurrentKafkaListenerContainerFactory concurrentKafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
}
Consumer Class:
@Slf4j
@Component
public class KafkaConsumer {
@KafkaListener(topics = "foo_topic", groupId = "group_1")
public void consume(String message) {
log.info(message);
}
}