I have difficulty understanding some Kafka concepts in Java Spring Boot. I’d like to test a consumer against a real Kafka broker running on a server, which has some producers that write / have already written data to various topics. I would like to establish a connection with the server, consume the data, and verify or process its content in a test.
An enormous majority of examples (actually all I have seen so far) in the internet refer to embedded kafka, EmbeddedKafkaBroker, and show both a producer and a consumer implemented on one machine, locally. I haven’t found any example that would explain how to make a connection with a remote kafka server and read data from a particular topic. I've written some code and I've printed the broker address with:
System.out.println(embeddedKafkaBroker.getBrokerAddress(0));
What I got is 127.0.0.1:9092, which means that it is local, so the connection with the remote server has not been established.
On the other hand, when I run the SpringBootApplication I get the payload from the remote broker.
Receiver:
@Component
public class Receiver {
private static final String TOPIC_NAME = "X";
private static final Logger LOGGER = LoggerFactory.getLogger(Receiver.class);
private CountDownLatch latch = new CountDownLatch(1);
public CountDownLatch getLatch() {
return latch;
}
@KafkaListener(topics = TOPIC_NAME)
public void receive(final byte[] payload) {
LOGGER.info("received the following payload: '{}'", payload);
latch.countDown();
}
}
Config:
@EnableKafka
@Configuration
public class ByteReceiverConfig {
@Autowired
EmbeddedKafkaBroker kafkaEmbeded;
@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;
@Value("${spring.kafka.consumer.group-id}")
private String groupIdConfig;
@Bean
public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory() {
final ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
@Bean
ConsumerFactory<Object, Object> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerProperties());
}
@Bean
Map<String, Object> consumerProperties() {
final Map<String, Object> properties =
KafkaTestUtils.consumerProps("junit-test", "true", this.kafkaEmbeded);
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupIdConfig);
return properties;
}
Test:
@EnableAutoConfiguration
@EnableKafka
@SpringBootTest(classes = {ByteReceiverConfig.class, Receiver.class})
@EmbeddedKafka
@ContextConfiguration(classes = ByteReceiverConfig.class)
@TestPropertySource(properties = { "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
"spring.kafka.consumer.group-id=EmbeddedKafkaTest"})
public class KafkaTest {
@Autowired
private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
@Autowired
EmbeddedKafkaBroker embeddedKafkaBroker;
@Autowired
Receiver receiver;
@BeforeEach
void waitForAssignment() {
for (MessageListenerContainer messageListenerContainer : kafkaListenerEndpointRegistry.getListenerContainers()) {
System.out.println(messageListenerContainer.getAssignedPartitions().isEmpty());
System.out.println(messageListenerContainer.toString());
System.out.println(embeddedKafkaBroker.getTopics().size());
System.out.println(embeddedKafkaBroker.getPartitionsPerTopic());
System.out.println(embeddedKafkaBroker.getBrokerAddress(0));
System.out.println(embeddedKafkaBroker.getBrokersAsString());
ContainerTestUtils.waitForAssignment(messageListenerContainer,
embeddedKafkaBroker.getPartitionsPerTopic());
}
@Test
public void testReceive() {
}
}
I would like somebody to shed some light on the following issues:
1.Can an instance of the class EmbeddedKafkaBroker be used to test data that comes from a remote broker, or is it only used for local tests, in which I would procude i.e send data to a topic that I created and consume data myself?
2.Is it possible to write a test class for a real kafka server? For instance to verify if a connection has been establish, or if a data has been read from a specific topic. What annotations, configurations, and classes would be needed in such case?
3.If I only want to consume data, do I have to provide the producer configuration in a config file (it would be strange, but all examples I have encountered so far did it)?
4.Do you know any resources (books, websites etc.) that show real examples of using kafka i.e. with a remote kafka server, with a procuder or a consumer only?
You don't need an embedded broker at all if you want to talk to an external broker only.
Yes, just set the bootstrap servers property appropriately.
No, you don't need producer configuration.
EDIT
@SpringBootApplication
public class So56044105Application {
public static void main(String[] args) {
SpringApplication.run(So56044105Application.class, args);
}
@Bean
public NewTopic topic() {
return new NewTopic("so56044105", 1, (short) 1);
}
}
spring.kafka.bootstrap-servers=10.0.0.8:9092
spring.kafka.consumer.enable-auto-commit=false
@RunWith(SpringRunner.class)
@SpringBootTest(classes = { So56044105Application.class, So56044105ApplicationTests.Config.class })
public class So56044105ApplicationTests {
@Autowired
public Config config;
@Test
public void test() throws InterruptedException {
assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(config.received.get(0)).isEqualTo("foo");
}
@Configuration
public static class Config implements ConsumerSeekAware {
List<String> received = new ArrayList<>();
CountDownLatch latch = new CountDownLatch(3);
@KafkaListener(id = "so56044105", topics = "so56044105")
public void listen(String in) {
System.out.println(in);
this.received.add(in);
this.latch.countDown();
}
@Override
public void registerSeekCallback(ConsumerSeekCallback callback) {
}
@Override
public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
System.out.println("Seeking to beginning");
assignments.keySet().forEach(tp -> callback.seekToBeginning(tp.topic(), tp.partition()));
}
@Override
public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
}
}
}