+++ /dev/null
-#!/bin/bash
-
-IMAGE=juplo/spring-consumer:1.1-SNAPSHOT
-
-if [ "$1" = "cleanup" ]
-then
- docker compose -f docker/docker-compose.yml down -t0 -v --remove-orphans
- mvn clean
- exit
-fi
-
-docker compose -f docker/docker-compose.yml up -d --remove-orphans kafka-1 kafka-2 kafka-3
-docker compose -f docker/docker-compose.yml rm -svf consumer
-
-if [[
- $(docker image ls -q $IMAGE) == "" ||
- "$1" = "build"
-]]
-then
- mvn clean install || exit
-else
- echo "Using image existing images:"
- docker image ls $IMAGE
-fi
-
-docker compose -f docker/docker-compose.yml up --remove-orphans setup || exit 1
-
-
-docker compose -f docker/docker-compose.yml up -d producer
-docker compose -f docker/docker-compose.yml up -d consumer
-
-sleep 5
-docker compose -f docker/docker-compose.yml stop consumer
-
-docker compose -f docker/docker-compose.yml start consumer
-sleep 5
-
-docker compose -f docker/docker-compose.yml stop producer consumer
-docker compose -f docker/docker-compose.yml logs consumer
try
{
log.info("{} - Subscribing to topic {}", id, topic);
- consumer.subscribe(Arrays.asList(topic));
+ // TODO: Aufruf von consumer.subscribe()
while (true)
{
- ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
-
- log.info("{} - Received {} messages", id, records.count());
- for (ConsumerRecord<String, String> record : records)
- {
- handleRecord(
- record.topic(),
- record.partition(),
- record.offset(),
- record.key(),
- record.value());
- }
+ // TODO: Poll & Consume aus Ihrer bisherigen Implementierung
}
}
catch(WakeupException e)
}
}
- private void handleRecord(
- String topic,
- Integer partition,
- Long offset,
- String key,
- String value)
- {
- consumed++;
- log.info("{} - partition={}-{}, offset={}: {}={}", id, topic, partition, offset, key, value);
- }
-
public void shutdown() throws InterruptedException
{