props.put("bootstrap.servers", bootstrapServer);
props.put("group.id", groupId);
props.put("client.id", id);
+ props.put("enable.auto.commit", false);
props.put("auto.offset.reset", autoOffsetReset);
props.put("metadata.max.age.ms", "1000");
props.put("key.deserializer", StringDeserializer.class.getName());
removed.getPartition(),
counter.getKey());
}
- repository.save(new StatisticsDocument(removed));
+ repository.save(new StatisticsDocument(removed, consumer.position(tp)));
});
}
partitions.forEach(tp ->
{
log.info("{} - adding partition: {}", id, tp);
- seen.put(
- tp,
+ StatisticsDocument document =
repository
.findById(tp.toString())
- .map(PartitionStatistics::new)
- .orElse(new PartitionStatistics(tp)));
+ .orElse(new StatisticsDocument(tp));
+ consumer.seek(tp, document.offset);
+ seen.put(tp, new PartitionStatistics(document));
});
}
});