dajac commented on code in PR #18014:
URL: https://github.com/apache/kafka/pull/18014#discussion_r1869101294
##########
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:
Here, I suggest to handle the known errors such as NOT_COORDINATOR,
COORDINATOR_LOADING, etc. As you optimistically send the write operations to
all the possible shards, you will get many "unknown" ones.
--
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
