From b3af1d816fdc1edd0ed6d1b0a6994995e5cb0d81 Mon Sep 17 00:00:00 2001
From: "Colin P. Mccabe" <cmccabe@confluent.io>
Date: Fri, 22 Mar 2019 15:31:42 -0700
Subject: [PATCH] KAFKA-8150: Fix bugs in handling null arrays in generated RPC
 code

ToString functions must not get a NullPointException.  read() functions
must properly translate a negative array length to a null field.
---
 .../apache/kafka/common/message/MessageTest.java  |  7 +++++--
 .../kafka/message/MessageDataGenerator.java       | 15 ++++++++++-----
 2 files changed, 15 insertions(+), 7 deletions(-)

diff --git a/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java b/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java
index 93a0930023af7..d573b3be31e13 100644
--- a/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java
@@ -38,7 +38,6 @@
 import org.apache.kafka.common.message.AddPartitionsToTxnRequestData.AddPartitionsToTxnTopic;
 import org.apache.kafka.common.message.AddPartitionsToTxnRequestData.AddPartitionsToTxnTopicSet;
 import org.junit.Assert;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.Timeout;
@@ -47,7 +46,6 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-@Ignore
 public final class MessageTest {
     @Rule
     final public Timeout globalTimeout = Timeout.millis(120000);
@@ -87,6 +85,11 @@ public void testRoundTrips() throws Exception {
             setHostFilter(null).
             setOperation((byte) 0).
             setPermissionType((byte) 0), (short) 0);
+        testMessageRoundTrips(new MetadataRequestData().
+            setTopics(null).
+            setAllowAutoTopicCreation(false).
+            setIncludeClusterAuthorizedOperations(false).
+            setIncludeTopicAuthorizedOperations(false));
     }
 
     private void testMessageRoundTrips(Message message) throws Exception {
diff --git a/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java b/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java
index 76029f4c1baf9..c8e70bba1e673 100644
--- a/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java
+++ b/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java
@@ -416,9 +416,8 @@ private void generateFieldReader(FieldSpec field, Versions curVersions) {
             buffer.printf("int arrayLength = readable.readInt();%n");
             buffer.printf("if (arrayLength < 0) {%n");
             buffer.incrementIndent();
-            buffer.printf("this.%s.clear(%s);%n",
-                field.camelCaseName(),
-                hasKeys ? "0" : "");
+            buffer.printf("this.%s = null;%n",
+                field.camelCaseName());
             buffer.decrementIndent();
             buffer.printf("} else {%n");
             buffer.incrementIndent();
@@ -1069,8 +1068,14 @@ private void generateFieldToString(String prefix, FieldSpec field) {
                 prefix, field.camelCaseName(), field.camelCaseName());
         } else if (field.type().isArray()) {
             headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
-            buffer.printf("+ \"%s%s=\" + MessageUtil.deepToString(%s.iterator())%n",
-                prefix, field.camelCaseName(), field.camelCaseName());
+            if (field.nullableVersions().empty()) {
+                buffer.printf("+ \"%s%s=\" + MessageUtil.deepToString(%s.iterator())%n",
+                    prefix, field.camelCaseName(), field.camelCaseName());
+            } else {
+                buffer.printf("+ \"%s%s=\" + ((%s == null) ? \"null\" : " +
+                    "MessageUtil.deepToString(%s.iterator()))%n",
+                    prefix, field.camelCaseName(), field.camelCaseName(), field.camelCaseName());
+            }
         } else {
             throw new RuntimeException("Unsupported field type " + field.type());
         }