props.put("bootstrap.servers", broker);
props.put("group.id", groupId); // ID für die Offset-Commits
props.put("client.id", clientId); // Nur zur Wiedererkennung
props.put("bootstrap.servers", broker);
props.put("group.id", groupId); // ID für die Offset-Commits
props.put("client.id", clientId); // Nur zur Wiedererkennung
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());
log.info("{} - Subscribing to topic test", id);
consumer.subscribe(Arrays.asList("test"));
log.info("{} - Subscribing to topic test", id);
consumer.subscribe(Arrays.asList("test"));
while (running)
{
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofSeconds(1));
while (running)
{
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofSeconds(1));
log.info("{} - Received {} messages", id, records.count());
for (ConsumerRecord<String, String> record : records)
{
log.info("{} - Received {} messages", id, records.count());
for (ConsumerRecord<String, String> record : records)
{