Shekharrajak commented on code in PR #22357:
URL: https://github.com/apache/kafka/pull/22357#discussion_r3430641740
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTransactionTest.java:
##########
@@ -665,4 +677,110 @@ private boolean hasDifferentLeader(Admin admin,
TopicPartition topicPartition, i
return false;
}
}
+
+ private void verifyLatestShareStateDeliveryState(
+ String groupId,
+ TopicIdPartition topicIdPartition,
+ long offset,
+ RecordState expectedState
+ ) throws InterruptedException {
+ int shareStatePartition = shareStatePartition(groupId,
topicIdPartition.topicId(), topicIdPartition.partition());
+ TopicPartition shareStateTopicPartition = new
TopicPartition(Topic.SHARE_GROUP_STATE_TOPIC_NAME, shareStatePartition);
+ ShareCoordinatorRecordSerde serde = new ShareCoordinatorRecordSerde();
+
+ try (Consumer<byte[], byte[]> consumer = cluster.consumer()) {
+ consumer.assign(List.of(shareStateTopicPartition));
+ TestUtils.waitForCondition(
+ () -> {
+ Byte actualState = latestShareStateDeliveryState(consumer,
serde, groupId, topicIdPartition, shareStateTopicPartition, offset);
+ return actualState != null && actualState ==
expectedState.id();
+ },
+ DEFAULT_MAX_WAIT_MS,
+ DEFAULT_POLL_INTERVAL_MS,
+ () -> "Timed out waiting for share state " + expectedState + "
for " + topicIdPartition);
+ }
+ }
+
+ private Byte latestShareStateDeliveryState(
+ Consumer<byte[], byte[]> consumer,
+ ShareCoordinatorRecordSerde serde,
+ String groupId,
+ TopicIdPartition topicIdPartition,
+ TopicPartition shareStateTopicPartition,
+ long offset
+ ) {
+ consumer.seekToBeginning(List.of(shareStateTopicPartition));
+ long endOffset =
consumer.endOffsets(List.of(shareStateTopicPartition)).get(shareStateTopicPartition);
+ Byte latestDeliveryState = null;
+
+ while (consumer.position(shareStateTopicPartition) < endOffset) {
+ ConsumerRecords<byte[], byte[]> records =
consumer.poll(Duration.ofMillis(500));
+ if (records.isEmpty()) {
+ break;
+ }
+
+ for (ConsumerRecord<byte[], byte[]> record :
records.records(shareStateTopicPartition)) {
+ CoordinatorRecord coordinatorRecord = serde.deserialize(
+ ByteBuffer.wrap(record.key()),
+ record.value() == null ? null :
ByteBuffer.wrap(record.value())
+ );
+ if (coordinatorRecord.value() == null ||
+ !matchesShareStateKey(coordinatorRecord, groupId,
topicIdPartition)) {
+ continue;
+ }
+
+ Object value = coordinatorRecord.value().message();
+ Byte deliveryState = null;
+ if (value instanceof ShareSnapshotValue snapshot) {
+ deliveryState =
snapshotDeliveryState(snapshot.stateBatches(), offset);
+ } else if (value instanceof ShareUpdateValue update) {
+ deliveryState = updateDeliveryState(update.stateBatches(),
offset);
+ }
+ if (deliveryState != null) {
+ latestDeliveryState = deliveryState;
+ }
+ }
+ }
+
+ return latestDeliveryState;
+ }
+
+ private boolean matchesShareStateKey(
+ CoordinatorRecord coordinatorRecord,
+ String groupId,
+ TopicIdPartition topicIdPartition
+ ) {
+ Object key = coordinatorRecord.key();
+ if (key instanceof ShareSnapshotKey snapshotKey) {
+ return snapshotKey.groupId().equals(groupId) &&
+ snapshotKey.topicId().equals(topicIdPartition.topicId()) &&
+ snapshotKey.partition() == topicIdPartition.partition();
+ }
+ if (key instanceof ShareUpdateKey updateKey) {
+ return updateKey.groupId().equals(groupId) &&
+ updateKey.topicId().equals(topicIdPartition.topicId()) &&
+ updateKey.partition() == topicIdPartition.partition();
+ }
+ return false;
+ }
+
+ private Byte snapshotDeliveryState(List<ShareSnapshotValue.StateBatch>
stateBatches, long offset) {
+ Byte deliveryState = null;
+ for (ShareSnapshotValue.StateBatch stateBatch : stateBatches) {
+ if (stateBatch.firstOffset() <= offset && offset <=
stateBatch.lastOffset()) {
Review Comment:
the offset delivery state
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
