depends_on:
- zookeeper
+ mongo:
+ image: mongo:4.4
+ ports:
+ - 27017:27017
+ environment:
+ MONGO_INITDB_ROOT_USERNAME: juplo
+ MONGO_INITDB_ROOT_PASSWORD: training
+
kafka-ui:
image: provectuslabs/kafka-ui:0.3.3
ports:
- 8081:8081
environment:
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
<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-actuator</artifactId>
@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");
EndlessConsumer consumer =
new EndlessConsumer(
Executors.newFixedThreadPool(1),
+ repository,
properties.getBootstrapServer(),
properties.getGroupId(),
properties.getClientId(),
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;
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<>()));
});
}
});
--- /dev/null
+package de.juplo.kafka;
+
+import org.springframework.data.mongodb.repository.MongoRepository;
+
+import java.util.Optional;
+
+
+public interface PartitionStatisticsRepository extends MongoRepository<StatisticsDocument, String>
+{
+ public Optional<StatisticsDocument> findById(String partition);
+}
--- /dev/null
+package de.juplo.kafka;
+
+import lombok.ToString;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.util.HashMap;
+import java.util.Map;
+
+
+@Document(collection = "statistics")
+@ToString
+public class StatisticsDocument
+{
+ @Id
+ public String id;
+ public Map<String, Integer> statistics;
+
+ public StatisticsDocument()
+ {
+ }
+
+ public StatisticsDocument(Integer partition, Map<String, Integer> statistics)
+ {
+ this.id = Integer.toString(partition);
+ this.statistics = statistics;
+ }
+}
web:
exposure:
include: "*"
+spring:
+ data:
+ mongodb:
+ uri: mongodb://juplo:training@localhost:27017
+ database: juplo
logging:
level:
root: INFO