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;
removed.getPartition(),
counter.getKey());
}
+ repository.save(new StatisticsDocument(removed));
});
}
partitions.forEach(tp ->
{
log.info("{} - adding partition: {}", id, tp);
- seen.put(tp, new PartitionStatistics(tp));
+ seen.put(
+ tp,
+ repository
+ .findById(tp.toString())
+ .map(PartitionStatistics::new)
+ .orElse(new PartitionStatistics(tp)));
});
}
});