dajac commented on code in PR #18695:
URL: https://github.com/apache/kafka/pull/18695#discussion_r2013553668
##########
tools/src/main/java/org/apache/kafka/tools/consumer/CoordinatorRecordMessageFormatter.java:
##########
@@ -31,44 +34,50 @@
import static java.nio.charset.StandardCharsets.UTF_8;
-public abstract class ApiMessageFormatter implements MessageFormatter {
-
+public abstract class CoordinatorRecordMessageFormatter implements
MessageFormatter {
private static final String TYPE = "type";
private static final String VERSION = "version";
private static final String DATA = "data";
private static final String KEY = "key";
private static final String VALUE = "value";
- static final String UNKNOWN = "unknown";
+
+ private final CoordinatorRecordSerde serde;
+
+ public CoordinatorRecordMessageFormatter(CoordinatorRecordSerde serde) {
+ this.serde = serde;
+ }
@Override
public void writeTo(ConsumerRecord<byte[], byte[]> consumerRecord,
PrintStream output) {
+ if (Objects.isNull(consumerRecord.key())) return;
+
ObjectNode json = new ObjectNode(JsonNodeFactory.instance);
+ try {
+ CoordinatorRecord record = serde.deserialize(
+ ByteBuffer.wrap(consumerRecord.key()),
+ consumerRecord.value() != null ?
ByteBuffer.wrap(consumerRecord.value()) : null
+ );
+
+ if (!isRecordTypeAllowed(record.key().apiKey())) return;
- byte[] key = consumerRecord.key();
- if (Objects.nonNull(key)) {
- short keyVersion = ByteBuffer.wrap(key).getShort();
- JsonNode dataNode = readToKeyJson(ByteBuffer.wrap(key));
+ json
+ .putObject(KEY)
+ .put(TYPE, record.key().apiKey())
Review Comment:
That's right. We will it in https://github.com/apache/kafka/pull/18510.
--
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]
