You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "smjn (via GitHub)" <gi...@apache.org> on 2025/04/11 08:23:44 UTC
[PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
smjn opened a new pull request, #19443:
URL: https://github.com/apache/kafka/pull/19443
* There could be scenarios where share partition records in `__share_group_state` internal topic have not been updated for a while implying these partitions are basically cold.
* In this situation, the presence of these basically holds back the pruner from keeping the topic clean and of manageable size.
* To remedy the situation, we have added a periodic `setupSnapshotColdPartitions` in `ShareCoordinatorService` which does a writeAll operation on the associated shards in the coordinator and forces snapshot creation for them. In this way the pruner can continue. This job has been added as a timer task.
* A new internal config `share.coordinator.cold.partition.snapshot.interval.ms` has been introduced to set the period of the job.
* Any failures are logged and ignored.
* New tests have been added to verify the feature.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #19443:
URL: https://github.com/apache/kafka/pull/19443#discussion_r2047117734
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -574,6 +575,46 @@ public CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecor
return new CoordinatorResult<>(List.of(record), responseData);
}
+ /**
+ * Iterates over the soft state to determine the share partitions whose last snapshot is
+ * older than the allowed time interval. The candidate share partitions are force snapshotted.
+ *
+ * @return A result containing snapshot records, if any, and a void response.
+ */
+ public CoordinatorResult<Void, CoordinatorRecord> snapshotColdPartitions() {
+ long coldSnapshottedPartitionsCount = shareStateMap.values().stream()
+ .filter(shareGroupOffset -> shareGroupOffset.createTimestamp() - shareGroupOffset.writeTimestamp() != 0)
+ .count();
+
+ // If all share partitions are snapshotted, it means that
+ // system is quiet and cold snapshotting will not help much.
+ if (coldSnapshottedPartitionsCount != 0 && coldSnapshottedPartitionsCount == shareStateMap.size()) {
Review Comment:
Incorporated
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #19443:
URL: https://github.com/apache/kafka/pull/19443#issuecomment-2812208506
When I delete the `__share_group_state` topic I get the following exception information:
```
[2025-04-17 09:41:17,524] INFO [ShareCoordinator id=1] Pruning records in __share_group_state-0 till offset 3. (org.apache.kafka.coordinator.share.ShareCoordinatorService)
[2025-04-17 09:41:17,527] ERROR [ShareCoordinator id=1] Received error in share-group state topic prune. (org.apache.kafka.coordinator.share.ShareCoordinatorService)
java.util.concurrent.CompletionException: org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.
at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:332) [?:?]
at java.base/java.util.concurrent.CompletableFuture.andTree(CompletableFuture.java:1527) [?:?]
at java.base/java.util.concurrent.CompletableFuture.allOf(CompletableFuture.java:2419) [?:?]
at org.apache.kafka.coordinator.share.ShareCoordinatorService$1.run(ShareCoordinatorService.java:281) [kafka-share-coordinator-4.1.0-SNAPSHOT.jar:?]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
at java.base/java.lang.Thread.run(Thread.java:840) [?:?]
Caused by: org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.
```
The background tasks perpetually throw exceptions in this situation, and I suspect that a more orderly leadership change could similarly make the code sad. While the coordinator runtime is properly able to handle unfortunate leadership events, I think the error handling of the background tasks in the share coordinator needs a little refinement.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19443:
URL: https://github.com/apache/kafka/pull/19443#discussion_r2048505136
##########
share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorShardTest.java:
##########
@@ -1357,6 +1358,350 @@ public void testInitializePartitionIdNonExistentInMetadataImage() {
verify(topicsImage, times(1)).getPartition(eq(TOPIC_ID), eq(0));
}
+ @Test
+ public void testSnapshotColdPartitionsNoEligiblePartitions() {
+ ShareCoordinatorShard shard = new ShareCoordinatorShardBuilder().build();
+ MetadataImage image = mock(MetadataImage.class);
+ shard.onNewMetadataImage(image, null);
+ int offset = 0;
+ int producerId = 0;
+ short producerEpoch = 0;
+ int leaderEpoch = 0;
+
+ long timestamp = TIME.milliseconds();
+
+ CoordinatorRecord record1 = CoordinatorRecord.record(
+ new ShareSnapshotKey()
+ .setGroupId(GROUP_ID)
+ .setTopicId(TOPIC_ID)
+ .setPartition(0),
+ new ApiMessageAndVersion(
+ new ShareSnapshotValue()
+ .setSnapshotEpoch(0)
+ .setStateEpoch(0)
+ .setLeaderEpoch(leaderEpoch)
+ .setCreateTimestamp(timestamp)
+ .setWriteTimestamp(timestamp)
+ .setStateBatches(List.of(
+ new ShareSnapshotValue.StateBatch()
+ .setFirstOffset(0)
+ .setLastOffset(10)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 0))),
+ (short) 0
+ )
+ );
+
+ CoordinatorRecord record2 = CoordinatorRecord.record(
+ new ShareSnapshotKey()
+ .setGroupId(GROUP_ID)
+ .setTopicId(TOPIC_ID)
+ .setPartition(1),
+ new ApiMessageAndVersion(
+ new ShareSnapshotValue()
+ .setSnapshotEpoch(0)
+ .setStateEpoch(0)
+ .setLeaderEpoch(leaderEpoch)
+ .setCreateTimestamp(timestamp)
+ .setWriteTimestamp(timestamp)
+ .setStateBatches(List.of(
+ new ShareSnapshotValue.StateBatch()
+ .setFirstOffset(0)
+ .setLastOffset(10)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 0))),
+ (short) 0
+ )
+ );
+
+ shard.replay(offset, producerId, producerEpoch, record1);
+ shard.replay(offset + 1, producerId, producerEpoch, record2);
+
+ assertNotNull(shard.getShareStateMapValue(SharePartitionKey.getInstance(GROUP_ID, TOPIC_ID, 0)));
+ assertNotNull(shard.getShareStateMapValue(SharePartitionKey.getInstance(GROUP_ID, TOPIC_ID, 1)));
+
+ TIME.sleep(5000); // Less than config.
+
+ assertEquals(0, shard.snapshotColdPartitions().records().size());
+ }
+
+ @Test
+ public void testSnapshotColdPartitionsSnapshotUpdateNotConsidered() {
+ ShareCoordinatorShard shard = new ShareCoordinatorShardBuilder().build();
+ MetadataImage image = mock(MetadataImage.class);
+ shard.onNewMetadataImage(image, null);
+ int offset = 0;
+ int producerId = 0;
+ short producerEpoch = 0;
+ int leaderEpoch = 0;
+
+ long timestamp = TIME.milliseconds();
+
+ CoordinatorRecord record1 = CoordinatorRecord.record(
+ new ShareSnapshotKey()
+ .setGroupId(GROUP_ID)
+ .setTopicId(TOPIC_ID)
+ .setPartition(0),
+ new ApiMessageAndVersion(
+ new ShareSnapshotValue()
+ .setSnapshotEpoch(0)
+ .setStateEpoch(0)
+ .setLeaderEpoch(leaderEpoch)
+ .setCreateTimestamp(timestamp)
+ .setWriteTimestamp(timestamp)
+ .setStateBatches(List.of(
+ new ShareSnapshotValue.StateBatch()
+ .setFirstOffset(0)
+ .setLastOffset(10)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 0))),
+ (short) 0
+ )
+ );
+
+ SharePartitionKey key = SharePartitionKey.getInstance(GROUP_ID, TOPIC_ID, 0);
+
+ shard.replay(offset, producerId, producerEpoch, record1);
+ assertNotNull(shard.getShareStateMapValue(key));
+
+ long sleep = 12000;
+ TIME.sleep(sleep);
+
+ List<CoordinatorRecord> expectedRecords = List.of(
+ CoordinatorRecord.record(
+ new ShareSnapshotKey()
+ .setGroupId(GROUP_ID)
+ .setTopicId(TOPIC_ID)
+ .setPartition(0),
+ new ApiMessageAndVersion(
+ new ShareSnapshotValue()
+ .setSnapshotEpoch(1)
+ .setStateEpoch(0)
+ .setLeaderEpoch(leaderEpoch)
+ .setCreateTimestamp(timestamp)
+ .setWriteTimestamp(timestamp + sleep)
+ .setStateBatches(List.of(
+ new ShareSnapshotValue.StateBatch()
+ .setFirstOffset(0)
+ .setLastOffset(10)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 0))),
+ (short) 0
+ )
+ )
+ );
+
+ assertEquals(expectedRecords, shard.snapshotColdPartitions().records());
+
+ shard.replay(offset + 1, producerId, producerEpoch, expectedRecords.get(0));
+ assertNotNull(shard.getShareStateMapValue(key));
+
+ CoordinatorRecord record2 = CoordinatorRecord.record(
+ new ShareUpdateKey()
+ .setGroupId(GROUP_ID)
+ .setTopicId(TOPIC_ID)
+ .setPartition(0),
+ new ApiMessageAndVersion(
+ new ShareUpdateValue()
+ .setSnapshotEpoch(0)
+ .setLeaderEpoch(leaderEpoch)
+ .setStateBatches(List.of(
+ new ShareUpdateValue.StateBatch()
+ .setFirstOffset(0)
+ .setLastOffset(10)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 0))),
+ (short) 0
+ )
+ );
+
+ shard.replay(offset + 2, producerId, producerEpoch, record2);
+
+ TIME.sleep(sleep);
+
+ assertNotNull(shard.getShareStateMapValue(key));
+ assertEquals(timestamp + sleep, shard.getShareStateMapValue(key).writeTimestamp()); // No snapshot since update has no time info.
+ }
+
+ @Test
+ public void testSnapshotColdPartitionsDoesNotPerpetuallySnapshot() {
+ ShareCoordinatorShard shard = new ShareCoordinatorShardBuilder().build();
+ MetadataImage image = mock(MetadataImage.class);
+ shard.onNewMetadataImage(image, null);
+ int offset = 0;
+ int producerId = 0;
+ short producerEpoch = 0;
+ int leaderEpoch = 0;
+
+ long timestamp = TIME.milliseconds();
+
+ CoordinatorRecord record1 = CoordinatorRecord.record(
+ new ShareSnapshotKey()
+ .setGroupId(GROUP_ID)
+ .setTopicId(TOPIC_ID)
+ .setPartition(0),
+ new ApiMessageAndVersion(
+ new ShareSnapshotValue()
+ .setSnapshotEpoch(0)
+ .setStateEpoch(0)
+ .setLeaderEpoch(leaderEpoch)
+ .setCreateTimestamp(timestamp)
+ .setWriteTimestamp(timestamp)
+ .setStateBatches(List.of(
+ new ShareSnapshotValue.StateBatch()
+ .setFirstOffset(0)
+ .setLastOffset(10)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 0))),
+ (short) 0
+ )
+ );
+
+ shard.replay(offset, producerId, producerEpoch, record1);
+ assertNotNull(shard.getShareStateMapValue(SharePartitionKey.getInstance(GROUP_ID, TOPIC_ID, 0)));
+
+ long sleep = 12000;
+ TIME.sleep(sleep);
+
+ List<CoordinatorRecord> expectedRecords = List.of(
+ CoordinatorRecord.record(
+ new ShareSnapshotKey()
+ .setGroupId(GROUP_ID)
+ .setTopicId(TOPIC_ID)
+ .setPartition(0),
+ new ApiMessageAndVersion(
+ new ShareSnapshotValue()
+ .setSnapshotEpoch(1)
+ .setStateEpoch(0)
+ .setLeaderEpoch(leaderEpoch)
+ .setCreateTimestamp(timestamp)
+ .setWriteTimestamp(timestamp + sleep)
+ .setStateBatches(List.of(
+ new ShareSnapshotValue.StateBatch()
+ .setFirstOffset(0)
+ .setLastOffset(10)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 0))),
+ (short) 0
+ )
+ )
+ );
+
+ assertEquals(expectedRecords, shard.snapshotColdPartitions().records());
+
+ shard.replay(offset + 1, producerId, producerEpoch, expectedRecords.get(0));
+ assertNotNull(shard.getShareStateMapValue(SharePartitionKey.getInstance(GROUP_ID, TOPIC_ID, 0)));
+
+ // Since all existing snapshots are already snapshotted, no new records will be created.
+ TIME.sleep(12000);
+
+ assertEquals(0, shard.snapshotColdPartitions().records().size());
+ }
+
+ @Test
+ public void testSnapshotColdPartitionsPartialEligiblePartitions() {
+ ShareCoordinatorShard shard = new ShareCoordinatorShardBuilder().build();
+ MetadataImage image = mock(MetadataImage.class);
+ shard.onNewMetadataImage(image, null);
+ int offset = 0;
+ int producerId = 0;
+ short producerEpoch = 0;
+ int leaderEpoch = 0;
+ SharePartitionKey key0 = SharePartitionKey.getInstance(GROUP_ID, TOPIC_ID, 0);
+ SharePartitionKey key1 = SharePartitionKey.getInstance(GROUP_ID, TOPIC_ID, 1);
+
+ long timestamp = TIME.milliseconds();
+ int record1SnapshotEpoch = 0;
+
+ CoordinatorRecord record1 = CoordinatorRecord.record(
+ new ShareSnapshotKey()
+ .setGroupId(GROUP_ID)
+ .setTopicId(TOPIC_ID)
+ .setPartition(0),
+ new ApiMessageAndVersion(
+ new ShareSnapshotValue()
+ .setSnapshotEpoch(record1SnapshotEpoch)
+ .setStateEpoch(0)
+ .setLeaderEpoch(leaderEpoch)
+ .setCreateTimestamp(timestamp)
+ .setWriteTimestamp(timestamp)
+ .setStateBatches(List.of(
+ new ShareSnapshotValue.StateBatch()
+ .setFirstOffset(0)
+ .setLastOffset(10)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 0))),
+ (short) 0
+ )
+ );
+
+ long delta = 15000; // 15 seconds
+
+ CoordinatorRecord record2 = CoordinatorRecord.record(
+ new ShareSnapshotKey()
+ .setGroupId(GROUP_ID)
+ .setTopicId(TOPIC_ID)
+ .setPartition(1),
+ new ApiMessageAndVersion(
+ new ShareSnapshotValue()
+ .setSnapshotEpoch(0)
+ .setStateEpoch(0)
+ .setLeaderEpoch(leaderEpoch)
+ .setCreateTimestamp(timestamp + delta)
+ .setWriteTimestamp(timestamp + delta)
+ .setStateBatches(List.of(
+ new ShareSnapshotValue.StateBatch()
+ .setFirstOffset(0)
+ .setLastOffset(10)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 0))),
+ (short) 0
+ )
+ );
+
+ shard.replay(offset, producerId, producerEpoch, record1);
+ shard.replay(offset + 1, producerId, producerEpoch, record2);
+
+ assertNotNull(shard.getShareStateMapValue(key0));
+ assertNotNull(shard.getShareStateMapValue(key1));
+ assertEquals(timestamp, shard.getShareStateMapValue(key0).writeTimestamp());
+ assertEquals(timestamp + delta, shard.getShareStateMapValue(key1).writeTimestamp());
+
+ long sleep = 12000;
+ TIME.sleep(sleep); // Record 1 is eligible now.
+
+ List<CoordinatorRecord> expectedRecords = List.of(
+ CoordinatorRecord.record(
+ new ShareSnapshotKey()
+ .setGroupId(GROUP_ID)
+ .setTopicId(TOPIC_ID)
+ .setPartition(0),
+ new ApiMessageAndVersion(
+ new ShareSnapshotValue()
+ .setSnapshotEpoch(record1SnapshotEpoch + 1)
+ .setStateEpoch(0)
+ .setLeaderEpoch(leaderEpoch)
+ .setCreateTimestamp(timestamp)
+ .setWriteTimestamp(timestamp + sleep)
+ .setStateBatches(List.of(
+ new ShareSnapshotValue.StateBatch()
+ .setFirstOffset(0)
+ .setLastOffset(10)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 0))),
+ (short) 0
+ )
+ )
+ );
+
+ List<CoordinatorRecord> records = shard.snapshotColdPartitions().records();
+ assertEquals(expectedRecords, records);
+
+ shard.replay(offset + 2, producerId, producerEpoch, records.get(0));
+
+ assertEquals(timestamp + delta, shard.getShareStateMapValue(key1).writeTimestamp());
+ }
Review Comment:
And I suppose you should `assertEquals(timestamp + sleep, shard.getShareStateMapValue(key0).writeTimestamp())` also.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #19443:
URL: https://github.com/apache/kafka/pull/19443#discussion_r2041964318
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareGroupOffset.java:
##########
@@ -245,4 +246,15 @@ public String toString() {
", stateBatches=" + stateBatches +
'}';
}
+
+ public Builder builderSupplier() {
+ return new Builder()
+ .setSnapshotEpoch(snapshotEpoch)
+ .setStateEpoch(stateEpoch)
+ .setLeaderEpoch(leaderEpoch)
+ .setStartOffset(startOffset)
+ .setStateBatches(new ArrayList<>(stateBatches))
Review Comment:
The idea was to prevent the caller from modifying the ShareGroupOffset pointed to by the `this`, while still being able to append additional batches to the resultant object.
If we don't duplicate.
The new SGO and the one pointed to by this will point to same internal statebatches list.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #19443:
URL: https://github.com/apache/kafka/pull/19443#issuecomment-2815838761
> When I delete the `__share_group_state` topic I get the following exception information:
>
> ```
> [2025-04-17 09:41:17,524] INFO [ShareCoordinator id=1] Pruning records in __share_group_state-0 till offset 3. (org.apache.kafka.coordinator.share.ShareCoordinatorService)
> [2025-04-17 09:41:17,527] ERROR [ShareCoordinator id=1] Received error in share-group state topic prune. (org.apache.kafka.coordinator.share.ShareCoordinatorService)
> java.util.concurrent.CompletionException: org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.
> at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:332) [?:?]
> at java.base/java.util.concurrent.CompletableFuture.andTree(CompletableFuture.java:1527) [?:?]
> at java.base/java.util.concurrent.CompletableFuture.allOf(CompletableFuture.java:2419) [?:?]
> at org.apache.kafka.coordinator.share.ShareCoordinatorService$1.run(ShareCoordinatorService.java:281) [kafka-share-coordinator-4.1.0-SNAPSHOT.jar:?]
> at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
> at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
> at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
> at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
> at java.base/java.lang.Thread.run(Thread.java:840) [?:?]
> Caused by: org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.
> ```
>
> The background tasks perpetually throw exceptions in this situation, and I suspect that a more orderly leadership change could similarly make the code sad. While the coordinator runtime is properly able to handle unfortunate leadership events, I think the error handling of the background tasks in the share coordinator needs a little refinement.
@AndrewJSchofield I do not understand this use case. The leadership changes are handled properly by the job as the topic partitions used by the job (activeTopicPartitions()) is maintained by the runtime.
If an internal TP is moved from broker 1 -> 2 then the corresponding active topic partition lists will be different for broker 1 and broker 2 (maintained by runtime). If you are talking about logging the exception - we can handle that in this case.
We have a similar test in `ShareConsumerTest.testShareConsumerAfterCoordinatorMovement` as well.
The exception is due to writing trying to delete the record offsets from the replicaManager, there is no exception being thrown from the runtime here.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19443:
URL: https://github.com/apache/kafka/pull/19443#discussion_r2041957517
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareGroupOffset.java:
##########
@@ -245,4 +246,15 @@ public String toString() {
", stateBatches=" + stateBatches +
'}';
}
+
+ public Builder builderSupplier() {
+ return new Builder()
+ .setSnapshotEpoch(snapshotEpoch)
+ .setStateEpoch(stateEpoch)
+ .setLeaderEpoch(leaderEpoch)
+ .setStartOffset(startOffset)
+ .setStateBatches(new ArrayList<>(stateBatches))
Review Comment:
@smjn, one question, why `stateBatches` needs to be wrapped with `ArrayList` here? Thank you.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #19443:
URL: https://github.com/apache/kafka/pull/19443#issuecomment-2802453814
> I've been playing with the code and generally it looks stable. However, I do observe that in a very quiet system, it will take snapshots every 5 minutes, even if that's all that it's doing. Surely, there comes a point where it's not worth writing a cold snapshot, because the amount of data to be replayed doesn't decrease at all by doing it.
We can modify the logic to stop cold snapshotting if all existing share partitions are already snapshotted. This means that after last snapshot they have not been written to organically.
so if we end up in state:
```
snap1-force
snap2-organic
snap3-force
```
cold snapshotting will happen at 5 minute intervals
but if we end up with
```
snap1-force
snap2-force
snap3-force
```
cold snapshotting will stop
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #19443:
URL: https://github.com/apache/kafka/pull/19443#issuecomment-2796503058
### Sample run
```
{"key":{"type":0,"data":{"groupId":"gs1","topicId":"usMKwc_eQ5214UuoShlXIg","partition":0}},"value":{"version":0,"data":{"snapshotEpoch":0,"stateEpoch":1,"leaderEpoch":-1,"startOffset":-1,"createTimestamp":1744356362818,"writeTimestamp":1744356362818,"stateBatches":[]}}}
{"key":{"type":1,"data":{"groupId":"gs1","topicId":"usMKwc_eQ5214UuoShlXIg","partition":0}},"value":{"version":0,"data":{"snapshotEpoch":0,"leaderEpoch":0,"startOffset":-1,"stateBatches":[]}}}
{"key":{"type":1,"data":{"groupId":"gs1","topicId":"usMKwc_eQ5214UuoShlXIg","partition":0}},"value":{"version":0,"data":{"snapshotEpoch":0,"leaderEpoch":0,"startOffset":16,"stateBatches":[{"firstOffset":16,"lastOffset":20,"deliveryState":2,"deliveryCount":1}]}}}
...
// Job run
{"key":{"type":0,"data":{"groupId":"gs1","topicId":"usMKwc_eQ5214UuoShlXIg","partition":0}},"value":{"version":0,"data":{"snapshotEpoch":1,"stateEpoch":1,"leaderEpoch":0,"startOffset":16,"createTimestamp":1744356362818,"writeTimestamp":1744356377744,"stateBatches":[{"firstOffset":16,"lastOffset":20,"deliveryState":2,"deliveryCount":1}]}}}
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #19443:
URL: https://github.com/apache/kafka/pull/19443#discussion_r2041964318
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareGroupOffset.java:
##########
@@ -245,4 +246,15 @@ public String toString() {
", stateBatches=" + stateBatches +
'}';
}
+
+ public Builder builderSupplier() {
+ return new Builder()
+ .setSnapshotEpoch(snapshotEpoch)
+ .setStateEpoch(stateEpoch)
+ .setLeaderEpoch(leaderEpoch)
+ .setStartOffset(startOffset)
+ .setStateBatches(new ArrayList<>(stateBatches))
Review Comment:
Otherwise they will point to the same internal state batches list.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #19443:
URL: https://github.com/apache/kafka/pull/19443
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #19443:
URL: https://github.com/apache/kafka/pull/19443#discussion_r2041964318
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareGroupOffset.java:
##########
@@ -245,4 +246,15 @@ public String toString() {
", stateBatches=" + stateBatches +
'}';
}
+
+ public Builder builderSupplier() {
+ return new Builder()
+ .setSnapshotEpoch(snapshotEpoch)
+ .setStateEpoch(stateEpoch)
+ .setLeaderEpoch(leaderEpoch)
+ .setStartOffset(startOffset)
+ .setStateBatches(new ArrayList<>(stateBatches))
Review Comment:
Otherwise they will point to the same internal state batches list. But I understand what you mean.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #19443:
URL: https://github.com/apache/kafka/pull/19443#discussion_r2041964318
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareGroupOffset.java:
##########
@@ -245,4 +246,15 @@ public String toString() {
", stateBatches=" + stateBatches +
'}';
}
+
+ public Builder builderSupplier() {
+ return new Builder()
+ .setSnapshotEpoch(snapshotEpoch)
+ .setStateEpoch(stateEpoch)
+ .setLeaderEpoch(leaderEpoch)
+ .setStartOffset(startOffset)
+ .setStateBatches(new ArrayList<>(stateBatches))
Review Comment:
Otherwise they will point to the same internal state batches list. But I understand what you mean, will close class for modification.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #19443:
URL: https://github.com/apache/kafka/pull/19443#discussion_r2041964318
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareGroupOffset.java:
##########
@@ -245,4 +246,15 @@ public String toString() {
", stateBatches=" + stateBatches +
'}';
}
+
+ public Builder builderSupplier() {
+ return new Builder()
+ .setSnapshotEpoch(snapshotEpoch)
+ .setStateEpoch(stateEpoch)
+ .setLeaderEpoch(leaderEpoch)
+ .setStartOffset(startOffset)
+ .setStateBatches(new ArrayList<>(stateBatches))
Review Comment:
The idea was to prevent the caller from modifying the ShareGroupOffset pointed to by the `this`, while still being able to append additional batches to the resultant object.
If we don't duplicate the new SGO and the one pointed to by this will point to same internal statebatches list.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19443:
URL: https://github.com/apache/kafka/pull/19443#discussion_r2047031680
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -574,6 +575,46 @@ public CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecor
return new CoordinatorResult<>(List.of(record), responseData);
}
+ /**
+ * Iterates over the soft state to determine the share partitions whose last snapshot is
+ * older than the allowed time interval. The candidate share partitions are force snapshotted.
+ *
+ * @return A result containing snapshot records, if any, and a void response.
+ */
+ public CoordinatorResult<Void, CoordinatorRecord> snapshotColdPartitions() {
+ long coldSnapshottedPartitionsCount = shareStateMap.values().stream()
+ .filter(shareGroupOffset -> shareGroupOffset.createTimestamp() - shareGroupOffset.writeTimestamp() != 0)
+ .count();
+
+ // If all share partitions are snapshotted, it means that
+ // system is quiet and cold snapshotting will not help much.
+ if (coldSnapshottedPartitionsCount != 0 && coldSnapshottedPartitionsCount == shareStateMap.size()) {
Review Comment:
I'm uncertain about this test. If the share state map is empty, then this if condition will be false, and we will proceed to do zero snapshots. Surely, this is a situation which should skip the snapshotting. What's wrong with just `coldSnapshottedPartitionsCount == shareStateMap.size()`? I know the code previously did that, but it seems a better test than what is there now.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18170: Add scheduled job to snapshot cold share partitions. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #19443:
URL: https://github.com/apache/kafka/pull/19443#issuecomment-2823884152
I have opened an issue to handle `__share_group_state` topic deletion (https://issues.apache.org/jira/browse/KAFKA-19189). In particular the record pruning throws an exception on every iteration without specifically handling `NOT_LEADER_OR_FOLLOWER`. Apart from that, this PR looks good to me.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org