Fix kwarg handing in kafka.protocol.struct.Struct (#1025)

2 files changed, 21 insertions, 2 deletions

diff --git a/kafka/protocol/struct.py b/kafka/protocol/struct.py
index 4c1afcb..3288172 100644
--- a/kafka/protocol/struct.py
+++ b/kafka/protocol/struct.py
@@ -18,7 +18,12 @@ class Struct(AbstractType):
         elif len(args) > 0:
             raise ValueError('Args must be empty or mirror schema')
         else:
-            self.__dict__.update(kwargs)
+            for name in self.SCHEMA.names:
+                self.__dict__[name] = kwargs.pop(name, None)
+            if kwargs:
+                raise ValueError('Keyword(s) not in schema %s: %s'
+                                 % (list(self.SCHEMA.names),
+                                    ', '.join(kwargs.keys())))
 
         # overloading encode() to support both class and instance
         # Without WeakMethod() this creates circular ref, which
diff --git a/test/test_protocol.py b/test/test_protocol.py
index 1c9f0f9..aa3dd17 100644
--- a/test/test_protocol.py
+++ b/test/test_protocol.py
@@ -7,8 +7,9 @@ import six
 
 from kafka.protocol.api import RequestHeader
 from kafka.protocol.commit import GroupCoordinatorRequest
-from kafka.protocol.fetch import FetchResponse
+from kafka.protocol.fetch import FetchRequest, FetchResponse
 from kafka.protocol.message import Message, MessageSet, PartialMessage
+from kafka.protocol.metadata import MetadataRequest
 from kafka.protocol.types import Int16, Int32, Int64, String
 
 
@@ -244,3 +245,16 @@ def test_decode_fetch_response_partial():
     m1 = partitions[0][3]
     assert len(m1) == 2
     assert m1[1] == (None, None, PartialMessage())
+
+
+def test_struct_unrecognized_kwargs():
+    try:
+        mr = MetadataRequest[0](topicz='foo')
+        assert False, 'Structs should not allow unrecognized kwargs'
+    except ValueError:
+        pass
+
+
+def test_struct_missing_kwargs():
+    fr = FetchRequest[0](max_wait_time=100)
+    assert fr.min_bytes is None