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("partition.assignment.strategy", "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
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"));
- // tag::loop[]
while (running)
{
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofSeconds(1));
- // Do something with the data...
- // end::loop[]
log.info("{} - Received {} messages", id, records.count());
for (ConsumerRecord<String, String> record : records)
{
record.value()
);
}
- // tag::loop[]
}
- // end::loop[]
}
catch(Exception e)
{