Fix compiled encode_into for None arrays (#2790)

Author: dpkp

kafka/protocol/new/schemas/fields/array.py

@@ -73,18 +73,19 @@ def emit_encode_into(self, ctx, val_expr, indent, version=None, compact=False, t
             an = ctx.next_var('an')
             ctx.emit(indent, '%s = len(%s) + 1 if %s is not None else 0' % (an, val_expr, val_expr))
             UnsignedVarInt32.emit_encode_into(ctx, an, indent)
+            ctx.emit(indent, 'if %s is not None:' % val_expr)
         else:
             ctx.emit(indent, 'if %s is None:' % val_expr)
             ctx.emit(indent, "    pack_into('>i', buf, pos, -1)")
             ctx.emit(indent, '    pos += 4')
             ctx.emit(indent, 'else:')
             ctx.emit(indent, "    pack_into('>i', buf, pos, len(%s))" % val_expr)
             ctx.emit(indent, '    pos += 4')
-        guard = indent + '    ' if not compact else indent
+        guard = indent + '    '
         item_var = ctx.next_var('ai')
         ctx.emit(guard, 'for %s in %s:' % (item_var, val_expr))
         self.array_of.emit_encode_into(ctx, item_var, guard + '    ',
-                                        version=version, compact=compact, tagged=tagged)
+                                       version=version, compact=compact, tagged=tagged)
 
     def emit_decode_from(self, ctx, var_name, indent, version=None, compact=False, tagged=False):
         n = ctx.next_var('n')

kafka/protocol/new/schemas/fields/struct_array.py

@@ -89,14 +89,15 @@ def emit_encode_into(self, ctx, val_expr, indent, version=None, compact=False, t
             an = ctx.next_var('an')
             ctx.emit(indent, '%s = len(%s) + 1 if %s is not None else 0' % (an, val_expr, val_expr))
             UnsignedVarInt32.emit_encode_into(ctx, an, indent)
+            ctx.emit(indent, 'if %s is not None:' % val_expr)
         else:
             ctx.emit(indent, 'if %s is None:' % val_expr)
             ctx.emit(indent, "    pack_into('>i', buf, pos, -1)")
             ctx.emit(indent, '    pos += 4')
             ctx.emit(indent, 'else:')
             ctx.emit(indent, "    pack_into('>i', buf, pos, len(%s))" % val_expr)
             ctx.emit(indent, '    pos += 4')
-        guard = indent + '    ' if not compact else indent
+        guard = indent + '    '
         item_var = ctx.next_var('si')
         if len(fields) == 1:
             # Single-field struct: items may be scalars (str, int, etc.)

test/protocol/new/metadata/test_new_metadata.py

@@ -28,6 +28,20 @@ def test_metadata_request_roundtrip(version):
     assert decoded == request
 
 
+@pytest.mark.parametrize("version", range(MetadataRequest.min_version, MetadataRequest.max_version + 1))
+def test_metadata_request_all_topics(version):
+    Topic = MetadataRequest.MetadataRequestTopic
+    request = MetadataRequest(
+        topics=None if version > 0 else [],
+        allow_auto_topic_creation=False if version >= 4 else True,
+        include_cluster_authorized_operations=True if 8 <= version <= 10 else False,
+        include_topic_authorized_operations=True if version >= 8 else False
+    )
+    encoded = request.encode(version=version)
+    decoded = MetadataRequest.decode(encoded, version=version)
+    assert decoded == request
+
+
 @pytest.mark.parametrize("version", range(MetadataResponse.min_version, MetadataResponse.max_version + 1))
 def test_metadata_response_roundtrip(version):
     Broker = MetadataResponse.MetadataResponseBroker