Skip to content

Commit b6adec4

Browse files
dajacdajac
authored
KAFKA-18616; Refactor Tools's ApiMessageFormatter (#18695)
This patch refactors the `ApiMessageFormatter` to follow what we have done in #18688. Reviewers: Chia-Ping Tsai <[email protected]>
1 parent 91758cc commit b6adec4

10 files changed

+508
-424
lines changed

‎checkstyle/import-control.xml

+1
Original file line numberDiff line numberDiff line change
@@ -338,6 +338,7 @@
338338

339339
<subpackage name="consumer">
340340
<allow pkg="org.apache.kafka.tools"/>
341+
<allow pkg="org.apache.kafka.coordinator.common.runtime" />
341342
<subpackage name="group">
342343
<allow pkg="org.apache.kafka.coordinator.group"/>
343344
<allow pkg="kafka.api"/>

‎tools/src/main/java/org/apache/kafka/tools/consumer/ApiMessageFormatter.java renamed to ‎tools/src/main/java/org/apache/kafka/tools/consumer/CoordinatorRecordMessageFormatter.java

+37-27
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,9 @@
1818

1919
import org.apache.kafka.clients.consumer.ConsumerRecord;
2020
import org.apache.kafka.common.MessageFormatter;
21+
import org.apache.kafka.common.protocol.ApiMessage;
22+
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
23+
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecordSerde;
2124

2225
import com.fasterxml.jackson.databind.JsonNode;
2326
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
@@ -31,44 +34,50 @@
3134

3235
import static java.nio.charset.StandardCharsets.UTF_8;
3336

34-
public abstract class ApiMessageFormatter implements MessageFormatter {
35-
37+
public abstract class CoordinatorRecordMessageFormatter implements MessageFormatter {
3638
private static final String TYPE = "type";
3739
private static final String VERSION = "version";
3840
private static final String DATA = "data";
3941
private static final String KEY = "key";
4042
private static final String VALUE = "value";
41-
static final String UNKNOWN = "unknown";
43+
44+
private final CoordinatorRecordSerde serde;
45+
46+
public CoordinatorRecordMessageFormatter(CoordinatorRecordSerde serde) {
47+
this.serde = serde;
48+
}
4249

4350
@Override
4451
public void writeTo(ConsumerRecord<byte[], byte[]> consumerRecord, PrintStream output) {
52+
if (Objects.isNull(consumerRecord.key())) return;
53+
4554
ObjectNode json = new ObjectNode(JsonNodeFactory.instance);
55+
try {
56+
CoordinatorRecord record = serde.deserialize(
57+
ByteBuffer.wrap(consumerRecord.key()),
58+
consumerRecord.value() != null ? ByteBuffer.wrap(consumerRecord.value()) : null
59+
);
60+
61+
if (!isRecordTypeAllowed(record.key().apiKey())) return;
4662

47-
byte[] key = consumerRecord.key();
48-
if (Objects.nonNull(key)) {
49-
short keyVersion = ByteBuffer.wrap(key).getShort();
50-
JsonNode dataNode = readToKeyJson(ByteBuffer.wrap(key));
63+
json
64+
.putObject(KEY)
65+
.put(TYPE, record.key().apiKey())
66+
.set(DATA, keyAsJson(record.key()));
5167

52-
if (dataNode instanceof NullNode) {
53-
return;
68+
if (Objects.nonNull(record.value())) {
69+
json
70+
.putObject(VALUE)
71+
.put(VERSION, record.value().version())
72+
.set(DATA, valueAsJson(record.value().message(), record.value().version()));
73+
} else {
74+
json.set(VALUE, NullNode.getInstance());
5475
}
55-
json.putObject(KEY)
56-
.put(TYPE, keyVersion)
57-
.set(DATA, dataNode);
58-
} else {
76+
} catch (CoordinatorRecordSerde.UnknownRecordTypeException ex) {
5977
return;
60-
}
61-
62-
byte[] value = consumerRecord.value();
63-
if (Objects.nonNull(value)) {
64-
short valueVersion = ByteBuffer.wrap(value).getShort();
65-
JsonNode dataNode = readToValueJson(ByteBuffer.wrap(value));
66-
67-
json.putObject(VALUE)
68-
.put(VERSION, valueVersion)
69-
.set(DATA, dataNode);
70-
} else {
71-
json.set(VALUE, NullNode.getInstance());
78+
} catch (RuntimeException ex) {
79+
throw new RuntimeException("Could not read record at offset " + consumerRecord.offset() +
80+
" due to: " + ex.getMessage(), ex);
7281
}
7382

7483
try {
@@ -78,6 +87,7 @@ public void writeTo(ConsumerRecord<byte[], byte[]> consumerRecord, PrintStream o
7887
}
7988
}
8089

81-
protected abstract JsonNode readToKeyJson(ByteBuffer byteBuffer);
82-
protected abstract JsonNode readToValueJson(ByteBuffer byteBuffer);
90+
protected abstract boolean isRecordTypeAllowed(short recordType);
91+
protected abstract JsonNode keyAsJson(ApiMessage message);
92+
protected abstract JsonNode valueAsJson(ApiMessage message, short version);
8393
}

‎tools/src/main/java/org/apache/kafka/tools/consumer/GroupMetadataMessageFormatter.java

+22-31
Original file line numberDiff line numberDiff line change
@@ -16,45 +16,36 @@
1616
*/
1717
package org.apache.kafka.tools.consumer;
1818

19-
import org.apache.kafka.common.errors.UnsupportedVersionException;
20-
import org.apache.kafka.common.protocol.ByteBufferAccessor;
19+
import org.apache.kafka.common.protocol.ApiMessage;
20+
import org.apache.kafka.coordinator.group.GroupCoordinatorRecordSerde;
21+
import org.apache.kafka.coordinator.group.generated.CoordinatorRecordJsonConverters;
2122
import org.apache.kafka.coordinator.group.generated.CoordinatorRecordType;
22-
import org.apache.kafka.coordinator.group.generated.GroupMetadataKey;
23-
import org.apache.kafka.coordinator.group.generated.GroupMetadataKeyJsonConverter;
24-
import org.apache.kafka.coordinator.group.generated.GroupMetadataValue;
25-
import org.apache.kafka.coordinator.group.generated.GroupMetadataValueJsonConverter;
2623

2724
import com.fasterxml.jackson.databind.JsonNode;
28-
import com.fasterxml.jackson.databind.node.NullNode;
29-
import com.fasterxml.jackson.databind.node.TextNode;
3025

31-
import java.nio.ByteBuffer;
26+
import java.util.Set;
27+
28+
public class GroupMetadataMessageFormatter extends CoordinatorRecordMessageFormatter {
29+
private static final Set<Short> ALLOWED_RECORDS = Set.of(
30+
CoordinatorRecordType.GROUP_METADATA.id()
31+
);
32+
33+
public GroupMetadataMessageFormatter() {
34+
super(new GroupCoordinatorRecordSerde());
35+
}
36+
37+
@Override
38+
protected boolean isRecordTypeAllowed(short recordType) {
39+
return ALLOWED_RECORDS.contains(recordType);
40+
}
3241

33-
public class GroupMetadataMessageFormatter extends ApiMessageFormatter {
3442
@Override
35-
protected JsonNode readToKeyJson(ByteBuffer byteBuffer) {
36-
try {
37-
switch (CoordinatorRecordType.fromId(byteBuffer.getShort())) {
38-
case GROUP_METADATA:
39-
return GroupMetadataKeyJsonConverter.write(
40-
new GroupMetadataKey(new ByteBufferAccessor(byteBuffer), (short) 0),
41-
(short) 0
42-
);
43-
44-
default:
45-
return NullNode.getInstance();
46-
}
47-
} catch (UnsupportedVersionException ex) {
48-
return NullNode.getInstance();
49-
}
43+
protected JsonNode keyAsJson(ApiMessage message) {
44+
return CoordinatorRecordJsonConverters.writeRecordKeyAsJson(message);
5045
}
5146

5247
@Override
53-
protected JsonNode readToValueJson(ByteBuffer byteBuffer) {
54-
short version = byteBuffer.getShort();
55-
if (version >= GroupMetadataValue.LOWEST_SUPPORTED_VERSION && version <= GroupMetadataValue.HIGHEST_SUPPORTED_VERSION) {
56-
return GroupMetadataValueJsonConverter.write(new GroupMetadataValue(new ByteBufferAccessor(byteBuffer), version), version);
57-
}
58-
return new TextNode(UNKNOWN);
48+
protected JsonNode valueAsJson(ApiMessage message, short version) {
49+
return CoordinatorRecordJsonConverters.writeRecordValueAsJson(message, version);
5950
}
6051
}

‎tools/src/main/java/org/apache/kafka/tools/consumer/OffsetsMessageFormatter.java

+22-32
Original file line numberDiff line numberDiff line change
@@ -16,50 +16,40 @@
1616
*/
1717
package org.apache.kafka.tools.consumer;
1818

19-
import org.apache.kafka.common.errors.UnsupportedVersionException;
20-
import org.apache.kafka.common.protocol.ByteBufferAccessor;
19+
import org.apache.kafka.common.protocol.ApiMessage;
20+
import org.apache.kafka.coordinator.group.GroupCoordinatorRecordSerde;
21+
import org.apache.kafka.coordinator.group.generated.CoordinatorRecordJsonConverters;
2122
import org.apache.kafka.coordinator.group.generated.CoordinatorRecordType;
22-
import org.apache.kafka.coordinator.group.generated.OffsetCommitKey;
23-
import org.apache.kafka.coordinator.group.generated.OffsetCommitKeyJsonConverter;
24-
import org.apache.kafka.coordinator.group.generated.OffsetCommitValue;
25-
import org.apache.kafka.coordinator.group.generated.OffsetCommitValueJsonConverter;
2623

2724
import com.fasterxml.jackson.databind.JsonNode;
28-
import com.fasterxml.jackson.databind.node.NullNode;
29-
import com.fasterxml.jackson.databind.node.TextNode;
3025

31-
import java.nio.ByteBuffer;
26+
import java.util.Set;
3227

3328
/**
3429
* Formatter for use with tools such as console consumer: Consumer should also set exclude.internal.topics to false.
3530
*/
36-
public class OffsetsMessageFormatter extends ApiMessageFormatter {
31+
public class OffsetsMessageFormatter extends CoordinatorRecordMessageFormatter {
32+
private static final Set<Short> ALLOWED_RECORDS = Set.of(
33+
CoordinatorRecordType.LEGACY_OFFSET_COMMIT.id(),
34+
CoordinatorRecordType.OFFSET_COMMIT.id()
35+
);
36+
37+
public OffsetsMessageFormatter() {
38+
super(new GroupCoordinatorRecordSerde());
39+
}
40+
3741
@Override
38-
protected JsonNode readToKeyJson(ByteBuffer byteBuffer) {
39-
try {
40-
switch (CoordinatorRecordType.fromId(byteBuffer.getShort())) {
41-
// We can read both record types with the offset commit one.
42-
case LEGACY_OFFSET_COMMIT:
43-
case OFFSET_COMMIT:
44-
return OffsetCommitKeyJsonConverter.write(
45-
new OffsetCommitKey(new ByteBufferAccessor(byteBuffer), (short) 0),
46-
(short) 0
47-
);
42+
protected boolean isRecordTypeAllowed(short recordType) {
43+
return ALLOWED_RECORDS.contains(recordType);
44+
}
4845

49-
default:
50-
return NullNode.getInstance();
51-
}
52-
} catch (UnsupportedVersionException ex) {
53-
return NullNode.getInstance();
54-
}
46+
@Override
47+
protected JsonNode keyAsJson(ApiMessage message) {
48+
return CoordinatorRecordJsonConverters.writeRecordKeyAsJson(message);
5549
}
5650

5751
@Override
58-
protected JsonNode readToValueJson(ByteBuffer byteBuffer) {
59-
short version = byteBuffer.getShort();
60-
if (version >= OffsetCommitValue.LOWEST_SUPPORTED_VERSION && version <= OffsetCommitValue.HIGHEST_SUPPORTED_VERSION) {
61-
return OffsetCommitValueJsonConverter.write(new OffsetCommitValue(new ByteBufferAccessor(byteBuffer), version), version);
62-
}
63-
return new TextNode(UNKNOWN);
52+
protected JsonNode valueAsJson(ApiMessage message, short version) {
53+
return CoordinatorRecordJsonConverters.writeRecordValueAsJson(message, version);
6454
}
6555
}

‎tools/src/main/java/org/apache/kafka/tools/consumer/TransactionLogMessageFormatter.java

+15-31
Original file line numberDiff line numberDiff line change
@@ -16,45 +16,29 @@
1616
*/
1717
package org.apache.kafka.tools.consumer;
1818

19-
import org.apache.kafka.common.errors.UnsupportedVersionException;
20-
import org.apache.kafka.common.protocol.ByteBufferAccessor;
21-
import org.apache.kafka.coordinator.transaction.generated.CoordinatorRecordType;
22-
import org.apache.kafka.coordinator.transaction.generated.TransactionLogKey;
23-
import org.apache.kafka.coordinator.transaction.generated.TransactionLogKeyJsonConverter;
24-
import org.apache.kafka.coordinator.transaction.generated.TransactionLogValue;
25-
import org.apache.kafka.coordinator.transaction.generated.TransactionLogValueJsonConverter;
19+
import org.apache.kafka.common.protocol.ApiMessage;
20+
import org.apache.kafka.coordinator.transaction.TransactionCoordinatorRecordSerde;
21+
import org.apache.kafka.coordinator.transaction.generated.CoordinatorRecordJsonConverters;
2622

2723
import com.fasterxml.jackson.databind.JsonNode;
28-
import com.fasterxml.jackson.databind.node.NullNode;
29-
import com.fasterxml.jackson.databind.node.TextNode;
3024

31-
import java.nio.ByteBuffer;
25+
public class TransactionLogMessageFormatter extends CoordinatorRecordMessageFormatter {
26+
public TransactionLogMessageFormatter() {
27+
super(new TransactionCoordinatorRecordSerde());
28+
}
3229

33-
public class TransactionLogMessageFormatter extends ApiMessageFormatter {
3430
@Override
35-
protected JsonNode readToKeyJson(ByteBuffer byteBuffer) {
36-
try {
37-
switch (CoordinatorRecordType.fromId(byteBuffer.getShort())) {
38-
case TRANSACTION_LOG:
39-
return TransactionLogKeyJsonConverter.write(
40-
new TransactionLogKey(new ByteBufferAccessor(byteBuffer), (short) 0),
41-
(short) 0
42-
);
31+
protected boolean isRecordTypeAllowed(short recordType) {
32+
return true;
33+
}
4334

44-
default:
45-
return NullNode.getInstance();
46-
}
47-
} catch (UnsupportedVersionException ex) {
48-
return NullNode.getInstance();
49-
}
35+
@Override
36+
protected JsonNode keyAsJson(ApiMessage message) {
37+
return CoordinatorRecordJsonConverters.writeRecordKeyAsJson(message);
5038
}
5139

5240
@Override
53-
protected JsonNode readToValueJson(ByteBuffer byteBuffer) {
54-
short version = byteBuffer.getShort();
55-
if (version >= TransactionLogValue.LOWEST_SUPPORTED_VERSION && version <= TransactionLogValue.HIGHEST_SUPPORTED_VERSION) {
56-
return TransactionLogValueJsonConverter.write(new TransactionLogValue(new ByteBufferAccessor(byteBuffer), version), version);
57-
}
58-
return new TextNode(UNKNOWN);
41+
protected JsonNode valueAsJson(ApiMessage message, short version) {
42+
return CoordinatorRecordJsonConverters.writeRecordValueAsJson(message, version);
5943
}
6044
}

‎tools/src/main/java/org/apache/kafka/tools/consumer/group/share/ShareGroupStateMessageFormatter.java

+2-1
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import org.apache.kafka.coordinator.share.generated.ShareUpdateKeyJsonConverter;
3131
import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
3232
import org.apache.kafka.coordinator.share.generated.ShareUpdateValueJsonConverter;
33+
import org.apache.kafka.tools.consumer.CoordinatorRecordMessageFormatter;
3334

3435
import com.fasterxml.jackson.databind.JsonNode;
3536
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
@@ -132,7 +133,7 @@ private JsonNode transferKeyMessageToJsonNode(ApiMessage logKey, short keyVersio
132133
* as per RPC spec.
133134
* To differentiate, we need to use the corresponding key versions. This is acceptable as
134135
* the records will always appear in pairs (key, value). However, this means that we cannot
135-
* extend {@link org.apache.kafka.tools.consumer.ApiMessageFormatter} as it requires overriding
136+
* extend {@link CoordinatorRecordMessageFormatter} as it requires overriding
136137
* readToValueJson whose signature does not allow for passing keyversion.
137138
*
138139
* @param byteBuffer - Represents the raw data read from the topic

‎tools/src/test/java/org/apache/kafka/tools/consumer/CoordinatorRecordMessageFormatterTest.java

+68
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.kafka.tools.consumer;
18+
19+
import org.apache.kafka.clients.consumer.ConsumerRecord;
20+
import org.apache.kafka.common.MessageFormatter;
21+
import org.apache.kafka.common.header.internals.RecordHeaders;
22+
import org.apache.kafka.common.record.TimestampType;
23+
24+
import org.junit.jupiter.api.TestInstance;
25+
import org.junit.jupiter.params.ParameterizedTest;
26+
import org.junit.jupiter.params.provider.Arguments;
27+
import org.junit.jupiter.params.provider.MethodSource;
28+
29+
import java.io.ByteArrayOutputStream;
30+
import java.io.PrintStream;
31+
import java.util.Optional;
32+
import java.util.stream.Stream;
33+
34+
import static java.util.Collections.emptyMap;
35+
import static org.junit.jupiter.api.Assertions.assertEquals;
36+
37+
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
38+
public abstract class CoordinatorRecordMessageFormatterTest {
39+
private static final String TOPIC = "TOPIC";
40+
41+
protected abstract CoordinatorRecordMessageFormatter formatter();
42+
protected abstract Stream<Arguments> parameters();
43+
44+
@ParameterizedTest
45+
@MethodSource("parameters")
46+
public void testMessageFormatter(byte[] keyBuffer, byte[] valueBuffer, String expectedOutput) {
47+
ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>(
48+
TOPIC,
49+
0,
50+
0,
51+
0L,
52+
TimestampType.CREATE_TIME,
53+
0,
54+
0,
55+
keyBuffer,
56+
valueBuffer,
57+
new RecordHeaders(),
58+
Optional.empty()
59+
);
60+
61+
try (MessageFormatter formatter = formatter()) {
62+
formatter.configure(emptyMap());
63+
ByteArrayOutputStream out = new ByteArrayOutputStream();
64+
formatter.writeTo(record, new PrintStream(out));
65+
assertEquals(expectedOutput.replaceAll("\\s+", ""), out.toString());
66+
}
67+
}
68+
}

0 commit comments

Comments
 (0)