import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
private final String topic;
private final KafkaConsumer<String, String> consumer;
private final String topic;
private final KafkaConsumer<String, String> consumer;
public SimpleConsumer(String broker, String topic, String groupId, String clientId)
{
public SimpleConsumer(String broker, String topic, String groupId, String clientId)
{
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"));
{
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofSeconds(1));
{
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)
{
log.info("{} - Closing the KafkaConsumer", id);
consumer.close();
log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
log.info("{} - Closing the KafkaConsumer", id);
consumer.close();
log.info("{}: Consumed {} messages in total, exiting!", id, consumed);