smjn commented on code in PR #18014:
URL: https://github.com/apache/kafka/pull/18014#discussion_r1869221577
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -240,9 +249,45 @@ public void startup(
log.info("Starting up.");
numPartitions = shareGroupTopicPartitionCount.getAsInt();
+ setupRecordPruning();
log.info("Startup complete.");
}
+ private void setupRecordPruning() {
+ timer.add(new TimerTask(config.shareCoordinatorTopicPruneIntervalMs())
{
+ @Override
+ public void run() {
+ for (int i = 0; i < numPartitions; i++) {
+ performRecordPruning(new
TopicPartition(Topic.SHARE_GROUP_STATE_TOPIC_NAME, i));
+ }
+ // perpetual recursion
+ setupRecordPruning();
+ }
+ });
+ }
+
+ private void performRecordPruning(TopicPartition tp) {
+ runtime.scheduleWriteOperation(
+ "write-state-record-prune",
+ tp,
+ Duration.ofMillis(config.shareCoordinatorWriteTimeoutMs()),
+ ShareCoordinatorShard::lastRedundantOffset
+ ).whenComplete((result, exception) -> {
+ if (exception != null) {
+ log.error("Last redundant offset lookup threw an error.",
exception);
+ return;
+ }
Review Comment:
I will remove the logging for above errors - they will happen in every case.
Since they timer job is periodic, we do not need any special handling anyway.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
