KAFKA-18409: ShareGroupStateMessageFormatter should use CoordinatorRecordMessageFormatter #18510

brandboat · 2025-01-13T16:22:31Z

ShareGroupStateMessageFormatter should extend CoordinatorRecordMessageFormatter in order to have a consistent handling of records of coordinators.

Reviewers: Ken Huang [email protected], David Jacot [email protected]

…matter

m1a2st

Thanks for the PR, left few comments. PTAL

m1a2st · 2025-01-13T16:51:13Z

tools/src/main/java/org/apache/kafka/tools/consumer/ApiMessageFormatter.java

@@ -38,7 +38,7 @@ public abstract class ApiMessageFormatter implements MessageFormatter {
    private static final String DATA = "data";
    private static final String KEY = "key";
    private static final String VALUE = "value";
-    static final String UNKNOWN = "unknown";
+    public static final String UNKNOWN = "unknown";


I think we should use the modifier protected.

Sure, thanks.

m1a2st · 2025-01-13T17:09:09Z

tools/src/main/java/org/apache/kafka/tools/consumer/ApiMessageFormatter.java

@@ -79,5 +79,5 @@ public void writeTo(ConsumerRecord<byte[], byte[]> consumerRecord, PrintStream o
    }

    protected abstract JsonNode readToKeyJson(ByteBuffer byteBuffer);
-    protected abstract JsonNode readToValueJson(ByteBuffer byteBuffer);
-}  
+    protected abstract JsonNode readToValueJson(ByteBuffer byteBuffer, short keyVersion);


Since the keyVersion parameter is only used by the ShareGroupStateMessageFormatter class, we should refactor this to make the method less abstract. Instead of requiring all subclasses to implement the keyVersion parameter, we could:

Move keyVersion out of the abstract definition

Provide a default implementation in the base class

WDYT?

Yes, I've thought about that before, but that means we need to add below methods to ApiMessageFormatter.java.

protected abstract JsonNode readToValueJson(ByteBuffer byteBuffer); protected JsonNode readToValueJson(ByteBuffer byteBuffer, short keyVersion) { return readToValueJson(byteBuffer); }

But doing that means I need to override two methods in ShareGroupStateMessageFormatter.java, particularly readToValueJson(ByteBuffer byteBuffer). This method will essentially be a dummy method, which feels somewhat awkward to implement, so I decided to add an extra parameter to it instead.

@Override protected JsonNode readToValueJson(ByteBuffer byteBuffer, short keyVersion) { short valueVersion = byteBuffer.getShort(); return readToSnapshotMessageValue(byteBuffer, keyVersion, valueVersion) .map(logValue -> transferValueMessageToJsonNode(logValue, valueVersion)) .orElseGet(() -> new TextNode(UNKNOWN)); } @Override protected JsonNode readToValueJson(ByteBuffer byteBuffer) { return null; }

Or you have other ideas? Let me know if you'd like any further adjustments!

I propose changing the JsonNode readToValueJson signature to accept a ConsumerRecord<byte[], byte[]>parameter. This would standardize the method signature across all subclasses. WDYT?

protected JsonNode readToValueJson(ConsumerRecord<byte[], byte[]> consumerRecord)

Not sure if this is a good idea, perhaps we can wait for others opinion.

key should not be converted to unknow json node

dajac · 2025-01-24T16:40:56Z

@brandboat What do you think about this one: #18695?

brandboat · 2025-01-26T19:22:48Z

@brandboat What do you think about this one: #18695?

I think this is a better approach. The entire refactor looks great to me. I'll hold off on this PR until #18695 is merged. Thank you for letting me know!

dajac · 2025-03-26T10:14:13Z

@brandboat We have merged #18695. We can proceed with your PR. Please ping me when the PR is ready for review.

brandboat · 2025-03-26T10:15:46Z

@brandboat We have merged #18695. We can proceed with your PR. Please ping me when the PR is ready for review.

Thanks for the heads-up, I'll handle this ASAP!

brandboat · 2025-03-28T16:27:40Z

Gentle ping @dajac, CI passed, could you take a look when you are available? Thank you!

dajac

@brandboat Thanks for the update. I left a few initial comments.

dajac · 2025-03-28T16:42:35Z

...st/java/org/apache/kafka/tools/consumer/group/share/ShareGroupStateMessageFormatterTest.java

@@ -137,23 +137,23 @@ private static Stream<Arguments> parameters() {
            Arguments.of(
                MessageUtil.toVersionPrefixedByteBuffer((short) 0, SHARE_SNAPSHOT_KEY).array(),
                MessageUtil.toVersionPrefixedByteBuffer((short) 0, SHARE_SNAPSHOT_VALUE).array(),
-                "{\"key\":{\"version\":0,\"data\":{\"groupId\":\"gs1\",\"topicId\":\"gtb2stGYRk-vWZ2zAozmoA\",\"partition\":0}},\"value\":{\"version\":0,\"data\":{\"snapshotEpoch\":0,\"stateEpoch\":1,\"leaderEpoch\":20,\"startOffset\":50,\"stateBatches\":[{\"firstOffset\":100,\"lastOffset\":200,\"deliveryState\":1,\"deliveryCount\":10},{\"firstOffset\":201,\"lastOffset\":210,\"deliveryState\":2,\"deliveryCount\":10}]}}}"
+                "{\"key\":{\"type\":0,\"data\":{\"groupId\":\"gs1\",\"topicId\":\"gtb2stGYRk-vWZ2zAozmoA\",\"partition\":0}},\"value\":{\"version\":0,\"data\":{\"snapshotEpoch\":0,\"stateEpoch\":1,\"leaderEpoch\":20,\"startOffset\":50,\"stateBatches\":[{\"firstOffset\":100,\"lastOffset\":200,\"deliveryState\":1,\"deliveryCount\":10},{\"firstOffset\":201,\"lastOffset\":210,\"deliveryState\":2,\"deliveryCount\":10}]}}}"


nit: Would it be possible to use the new test blocks like we did for the other formatters?

dajac · 2025-03-28T16:46:45Z

...st/java/org/apache/kafka/tools/consumer/group/share/ShareGroupStateMessageFormatterTest.java

-                new RuntimeException("non-nullable field stateBatches was serialized as null")
+                new RuntimeException("Could not read record at offset 0 due to: " +
+                        "Could not read record with version 0 from value's buffer due to: " +
+                        "non-nullable field stateBatches was serialized as null.")
            )
        );
    }


Could we remove testShareGroupStateMessageFormatter? We have it in the parent class now.

Thank you! I've addressed all the comments!

dajac · 2025-03-31T09:43:05Z

@brandboat It looks like https://github.com/apache/kafka/pull/18510/files#r2019007145 has not been addressed yet. Could you please address it?

brandboat · 2025-03-31T10:30:20Z

@brandboat It looks like https://github.com/apache/kafka/pull/18510/files#r2019007145 has not been addressed yet. Could you please address it?

My apologies, I've replaced all long string in the test with text blocks.

dajac

lgtm, thanks

…cordMessageFormatter (apache#18510) ShareGroupStateMessageFormatter should extend CoordinatorRecordMessageFormatter in order to have a consistent handling of records of coordinators. Reviewers: Ken Huang <[email protected]>, David Jacot <[email protected]>

KAFKA-18409: ShareGroupStateMessageFormatter should use ApiMessageFor…

a10f245

…matter

github-actions bot added the tools label Jan 13, 2025

m1a2st reviewed Jan 13, 2025

View reviewed changes

brandboat added 4 commits January 16, 2025 20:44

Merge remote-tracking branch 'origin/trunk' into KAFKA-18409

514cb49

Address comment

13bb5ef

Fix transferKeyMessageToJsonNode

aeea284

key should not be converted to unknow json node

Merge remote-tracking branch 'origin/trunk' into KAFKA-18409

62f4d86

brandboat changed the title ~~KAFKA-18409: ShareGroupStateMessageFormatter should use ApiMessageFormatter~~ (WIP) KAFKA-18409: ShareGroupStateMessageFormatter should use ApiMessageFormatter Jan 26, 2025

dajac mentioned this pull request Mar 26, 2025

KAFKA-18616; Refactor Tools's ApiMessageFormatter #18695

Merged

brandboat added 2 commits March 28, 2025 11:32

Merge remote-tracking branch 'origin/trunk' into KAFKA-18409

94d2b55

update

5dc919b

brandboat changed the title ~~(WIP) KAFKA-18409: ShareGroupStateMessageFormatter should use ApiMessageFormatter~~ KAFKA-18409: ShareGroupStateMessageFormatter should use ApiMessageFormatter Mar 28, 2025

brandboat changed the title ~~KAFKA-18409: ShareGroupStateMessageFormatter should use ApiMessageFormatter~~ KAFKA-18409: ShareGroupStateMessageFormatter should use CoordinatorRecordMessageFormatter Mar 28, 2025

dajac reviewed Mar 28, 2025

View reviewed changes

brandboat added 3 commits March 29, 2025 10:18

address comments

12672f8

Merge remote-tracking branch 'origin/trunk' into KAFKA-18409

1e8f15b

fix import order

948d2b7

brandboat added 2 commits March 31, 2025 18:26

use new text blocks

7e6f2c3

Merge remote-tracking branch 'origin/trunk' into KAFKA-18409

270661f

dajac approved these changes Mar 31, 2025

View reviewed changes

dajac merged commit 5982943 into apache:trunk Mar 31, 2025
23 checks passed

brandboat deleted the KAFKA-18409 branch March 31, 2025 15:00

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18409: ShareGroupStateMessageFormatter should use CoordinatorRecordMessageFormatter #18510

KAFKA-18409: ShareGroupStateMessageFormatter should use CoordinatorRecordMessageFormatter #18510

brandboat commented Jan 13, 2025 •

edited by dajac

Loading

m1a2st left a comment

m1a2st Jan 13, 2025

brandboat Jan 14, 2025

m1a2st Jan 13, 2025

brandboat Jan 14, 2025

m1a2st Jan 14, 2025

brandboat Jan 14, 2025

dajac commented Jan 24, 2025

brandboat commented Jan 26, 2025

dajac commented Mar 26, 2025

brandboat commented Mar 26, 2025

brandboat commented Mar 28, 2025

dajac left a comment

dajac Mar 28, 2025

dajac Mar 28, 2025

brandboat Mar 29, 2025

dajac commented Mar 31, 2025

brandboat commented Mar 31, 2025

dajac left a comment

KAFKA-18409: ShareGroupStateMessageFormatter should use CoordinatorRecordMessageFormatter #18510

KAFKA-18409: ShareGroupStateMessageFormatter should use CoordinatorRecordMessageFormatter #18510

Conversation

brandboat commented Jan 13, 2025 • edited by dajac Loading

m1a2st left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dajac commented Jan 24, 2025

brandboat commented Jan 26, 2025

dajac commented Mar 26, 2025

brandboat commented Mar 26, 2025

brandboat commented Mar 28, 2025

dajac left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dajac commented Mar 31, 2025

brandboat commented Mar 31, 2025

dajac left a comment

Choose a reason for hiding this comment

brandboat commented Jan 13, 2025 •

edited by dajac

Loading