version: '3.2'
services:
zookeeper:
- image: confluentinc/cp-zookeeper:7.0.2
+ image: confluentinc/cp-zookeeper:7.1.3
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ports:
- 2181:2181
kafka:
- image: confluentinc/cp-kafka:7.0.2
+ image: confluentinc/cp-kafka:7.1.3
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
depends_on:
- zookeeper
+ mongo:
+ image: mongo:4.4
+ ports:
+ - 27017:27017
+ environment:
+ MONGO_INITDB_ROOT_USERNAME: juplo
+ MONGO_INITDB_ROOT_PASSWORD: training
+
+ express:
+ image: mongo-express
+ ports:
+ - 8090:8081
+ environment:
+ ME_CONFIG_MONGODB_ADMINUSERNAME: juplo
+ ME_CONFIG_MONGODB_ADMINPASSWORD: training
+ ME_CONFIG_MONGODB_URL: mongodb://juplo:training@mongo:27017/
+
kafka-ui:
image: provectuslabs/kafka-ui:0.3.3
ports:
producer:
image: juplo/endless-producer:1.0-SNAPSHOT
ports:
- - 8000:8080
+ - 8080:8080
environment:
+ server.port: 8080
producer.bootstrap-server: kafka:9092
producer.client-id: producer
producer.topic: test
consumer:
image: juplo/endless-consumer:1.0-SNAPSHOT
ports:
- - 8081:8081
+ - 8081:8080
environment:
+ server.port: 8080
consumer.bootstrap-server: kafka:9092
- consumer.client-id: my-group
consumer.client-id: consumer
consumer.topic: test
+ spring.data.mongodb.uri: mongodb://juplo:training@mongo:27017
+ spring.data.mongodb.database: juplo
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
- <version>2.6.5</version>
+ <version>2.7.2</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-data-mongodb</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-validation</artifactId>
+ </dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>build-info</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>pl.project13.maven</groupId>
+ <artifactId>git-commit-id-plugin</artifactId>
</plugin>
<plugin>
<groupId>io.fabric8</groupId>
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
- import org.springframework.util.Assert;
import java.util.concurrent.Executors;
@Bean
- public EndlessConsumer consumer()
+ public EndlessConsumer consumer(PartitionStatisticsRepository repository)
{
- Assert.hasText(properties.getBootstrapServer(), "consumer.bootstrap-server must be set");
- Assert.hasText(properties.getGroupId(), "consumer.group-id must be set");
- Assert.hasText(properties.getClientId(), "consumer.client-id must be set");
- Assert.hasText(properties.getTopic(), "consumer.topic must be set");
-
EndlessConsumer consumer =
new EndlessConsumer(
Executors.newFixedThreadPool(1),
+ repository,
properties.getBootstrapServer(),
properties.getGroupId(),
properties.getClientId(),
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
- import java.util.concurrent.Future;
- import java.util.concurrent.atomic.AtomicBoolean;
+ import java.util.concurrent.locks.Condition;
+ import java.util.concurrent.locks.Lock;
+ import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class EndlessConsumer implements Runnable
{
private final ExecutorService executor;
+ private final PartitionStatisticsRepository repository;
private final String bootstrapServer;
private final String groupId;
private final String id;
private final String topic;
private final String autoOffsetReset;
- private AtomicBoolean running = new AtomicBoolean();
+ private final Lock lock = new ReentrantLock();
+ private final Condition condition = lock.newCondition();
+ private boolean running = false;
+ private Exception exception;
private long consumed = 0;
private KafkaConsumer<String, String> consumer = null;
- private Future<?> future = null;
+
private final Map<Integer, Map<String, Integer>> seen = new HashMap<>();
public EndlessConsumer(
ExecutorService executor,
+ PartitionStatisticsRepository repository,
String bootstrapServer,
String groupId,
String clientId,
String autoOffsetReset)
{
this.executor = executor;
+ this.repository = repository;
this.bootstrapServer = bootstrapServer;
this.groupId = groupId;
this.id = clientId;
tp.partition(),
key);
}
+ repository.save(new StatisticsDocument(tp.partition(), removed));
});
}
partitions.forEach(tp ->
{
log.info("{} - adding partition: {}", id, tp);
- seen.put(tp.partition(), new HashMap<>());
+ seen.put(
+ tp.partition(),
+ repository
+ .findById(Integer.toString(tp.partition()))
+ .map(document -> document.statistics)
+ .orElse(new HashMap<>()));
});
}
});
catch(WakeupException e)
{
log.info("{} - RIIING!", id);
+ shutdown();
}
catch(Exception e)
{
log.error("{} - Unexpected error: {}", id, e.toString(), e);
- running.set(false); // Mark the instance as not running
+ shutdown(e);
}
finally
{
}
}
+ private void shutdown()
+ {
+ shutdown(null);
+ }
+
+ private void shutdown(Exception e)
+ {
+ lock.lock();
+ try
+ {
+ running = false;
+ exception = e;
+ condition.signal();
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
public Map<Integer, Map<String, Integer>> getSeen()
{
return seen;
}
- public synchronized void start()
+ public void start()
{
- boolean stateChanged = running.compareAndSet(false, true);
- if (!stateChanged)
- throw new RuntimeException("Consumer instance " + id + " is already running!");
+ lock.lock();
+ try
+ {
+ if (running)
+ throw new IllegalStateException("Consumer instance " + id + " is already running!");
- log.info("{} - Starting - consumed {} messages before", id, consumed);
- future = executor.submit(this);
+ log.info("{} - Starting - consumed {} messages before", id, consumed);
+ running = true;
+ exception = null;
+ executor.submit(this);
+ }
+ finally
+ {
+ lock.unlock();
+ }
}
public synchronized void stop() throws ExecutionException, InterruptedException
{
- boolean stateChanged = running.compareAndSet(true, false);
- if (!stateChanged)
- throw new RuntimeException("Consumer instance " + id + " is not running!");
-
- log.info("{} - Stopping", id);
- consumer.wakeup();
- future.get();
- log.info("{} - Stopped - consumed {} messages so far", id, consumed);
+ lock.lock();
+ try
+ {
+ if (!running)
+ throw new IllegalStateException("Consumer instance " + id + " is not running!");
+
+ log.info("{} - Stopping", id);
+ consumer.wakeup();
+ condition.await();
+ log.info("{} - Stopped - consumed {} messages so far", id, consumed);
+ }
+ finally
+ {
+ lock.unlock();
+ }
}
@PreDestroy
{
log.info("{} - Was already stopped", id);
}
+ catch (Exception e)
+ {
+ log.error("{} - Unexpected exception while trying to stop the consumer", id, e);
+ }
finally
{
log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
}
}
+
+ public boolean running()
+ {
+ lock.lock();
+ try
+ {
+ return running;
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
+ public Optional<Exception> exitStatus()
+ {
+ lock.lock();
+ try
+ {
+ if (running)
+ throw new IllegalStateException("No exit-status available: Consumer instance " + id + " is running!");
+
+ return Optional.ofNullable(exception);
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
}
projects / demos / kafka / training / commitdiff