Manual json schema edits to restore fields dropped in Apache 4.0

kafka/protocol/new/schemas/resources/DescribeConfigsResponse.json

@@ -46,6 +46,8 @@
           "about": "True if the configuration is read-only." },
         { "name": "ConfigSource", "type": "int8", "versions": "1+", "default": "-1", "ignorable": true,
           "about": "The configuration source." },
+        { "name": "IsDefault", "type": "bool", "versions": "0",
+          "about": "True if the configuration is not set." },
         { "name": "IsSensitive", "type": "bool", "versions": "0+",
           "about": "True if this configuration is sensitive." },
         { "name": "Synonyms", "type": "[]DescribeConfigsSynonym", "versions": "1+", "ignorable": true,

kafka/protocol/new/schemas/resources/ListOffsetsRequest.json

@@ -59,7 +59,9 @@
         { "name": "CurrentLeaderEpoch", "type": "int32", "versions": "4+", "default": "-1", "ignorable": true,
           "about": "The current leader epoch." },
         { "name": "Timestamp", "type": "int64", "versions": "0+",
-          "about": "The current timestamp." }
+          "about": "The current timestamp." },
+        { "name": "MaxNumOffsets", "type": "int32", "versions": "0", "default": "1",
+          "about": "The maximum number of offsets to report." }
       ]}
     ]},
     { "name": "TimeoutMs", "type": "int32", "versions": "10+", "ignorable": true,

kafka/protocol/new/schemas/resources/ListOffsetsResponse.json

@@ -55,6 +55,8 @@
           "about": "The partition index." },
         { "name": "ErrorCode", "type": "int16", "versions": "0+",
           "about": "The partition error code, or 0 if there was no error." },
+        { "name": "OldStyleOffsets", "type": "[]int64", "versions": "0", "ignorable": false,
+          "about": "The result offsets." },
         { "name": "Timestamp", "type": "int64", "versions": "1+", "default": "-1", "ignorable": false,
           "about": "The timestamp associated with the returned offset." },
         { "name": "Offset", "type": "int64", "versions": "1+", "default": "-1", "ignorable": false,

test/protocol/new/admin/test_new_admin.py

@@ -103,14 +103,48 @@ def test_describe_configs_request_roundtrip(version):
             configuration_keys=["cleanup.policy"]
         )
     ]
-    data = DescribeConfigsRequest(
+    request = DescribeConfigsRequest(
         resources=resources,
         include_synonyms=True if version >= 1 else False,
         include_documentation=True if version >= 3 else False
     )
-    encoded = DescribeConfigsRequest.encode(data, version=version)
+    encoded = request.encode(version=version)
     decoded = DescribeConfigsRequest.decode(encoded, version=version)
-    assert decoded == data
+    assert decoded == request
+
+    Result = DescribeConfigsResponse.DescribeConfigsResult
+    Config = Result.DescribeConfigsResourceResult
+    Synonym = Config.DescribeConfigsSynonym
+    response = DescribeConfigsResponse(
+        throttle_time_ms=0,
+        results=[
+            DescribeConfigsResponse.DescribeConfigsResult(
+                error_code=0,
+                error_message=None,
+                resource_type=2,
+                resource_name="test-topic",
+                configs=[
+                    Config(
+                        name='foo',
+                        value='bar',
+                        read_only=False,
+                        config_source=2 if version >= 1 else -1,
+                        is_default=True if version == 0 else False,
+                        is_sensitive=False,
+                        synonyms=[
+                            Synonym(name='fizz', value='buzz', source=2)
+                        ] if version >= 1 else [],
+                        config_type=1 if version >=3 else 0,
+                        documentation='whats up doc' if version >=3 else '',
+                    )
+                ]
+            )
+        ],
+    )
+
+    encoded = DescribeConfigsResponse.encode(response, version=version)
+    decoded = DescribeConfigsResponse.decode(encoded, version=version)
+    assert decoded == response
 
 
 @pytest.mark.parametrize("version", range(CreateAclsRequest.min_version, CreateAclsRequest.max_version + 1))

test/protocol/new/consumer/test_new_list_offsets.py

@@ -11,10 +11,11 @@ def test_list_offsets_request_roundtrip(version):
             partitions=[
                 ListOffsetsRequest.ListOffsetsTopic.ListOffsetsPartition(
                     partition_index=0,
-                    current_leader_epoch=-1 if version >= 9 else -1, # default is -1
-                    timestamp=1000
+                    current_leader_epoch=1 if version >= 9 else -1,
+                    timestamp=1000,
+                    max_num_offsets=2 if version == 0 else 1
                 )
-            ]
+            ],
         )
     ]
     data = ListOffsetsRequest(
@@ -37,11 +38,12 @@ def test_list_offsets_response_roundtrip(version):
                 ListOffsetsResponse.ListOffsetsTopicResponse.ListOffsetsPartitionResponse(
                     partition_index=0,
                     error_code=0,
+                    old_style_offsets=[0] if version == 0 else [],
                     timestamp=1000 if version >= 1 else -1,
                     offset=1000 if version >= 1 else -1,
                     leader_epoch=1 if version >= 4 else -1
                 )
-            ]
+            ],
         )
     ]
     data = ListOffsetsResponse(