Merge branch '7.0.x' into 7.1.x

confluentinc · Jun 30, 2023 · 49f93a4 · 49f93a4
2 parents 1499fed + a73e371
commit 49f93a4
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 2 deletions.
diff --git a/avro-serializer/src/test/java/io/confluent/kafka/formatter/AvroMessageFormatterTest.java b/avro-serializer/src/test/java/io/confluent/kafka/formatter/AvroMessageFormatterTest.java
@@ -129,6 +129,11 @@ public void testDeserializeRecordWithKeyAndValueAndPrintingKeyAndSchemaIds()
     PrintStream ps = new PrintStream(baos);
     formatter.writeTo(recordWithKeyAndValue, ps);
     assertEquals("\"" + KEY_STRING + "\"\t2\t\"" + A_STRING + "\"\t1\n", baos.toString());
+
+    baos = new ByteArrayOutputStream();
+    ps = new PrintStream(baos);
+    formatter.writeTo(recordWithValue, ps);
+    assertEquals("null\tnull\t\"" + A_STRING + "\"\t1\n", baos.toString());
   }
 
   @Test

diff --git a/schema-serializer/src/main/java/io/confluent/kafka/formatter/SchemaMessageFormatter.java b/schema-serializer/src/main/java/io/confluent/kafka/formatter/SchemaMessageFormatter.java
@@ -169,8 +169,12 @@ public void writeTo(ConsumerRecord<byte[], byte[]> consumerRecord, PrintStream o
         }
         if (printKeyId) {
           output.write(idSeparator);
-          int schemaId = schemaIdFor(consumerRecord.key());
-          output.print(schemaId);
+          if (consumerRecord.key() != null) {
+            int schemaId = schemaIdFor(consumerRecord.key());
+            output.print(schemaId);
+          } else {
+            output.write(NULL_BYTES);
+          }
         }
         output.write(keySeparator);
       } catch (IOException ioe) {