Repository: dpkp/kafka-python

Issue:
Support for transactional messages
The client doesn't support transactional messages. On trying to consume messages created by a transactional producer (scala/java based) the python client fails to correctly give values for key/value.

Hints from issue comments:
@jutley: We've been having issues with one of our consumers that uses this library, and realized that it's because of transactional messages. In particular, we're seeing that the control messages are being exposed to the client, which should not happen. See the documentation [here](https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol), which says:
> Clients should not return control batches (ie. those with this bit set) to applications. (since 0.11.0.0)@jeffwidman: A PR would be most welcome!@jouve: https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol


  > The sixth lowest bit indicates whether the RecordBatch includes a control message. 1 indicates that the RecordBatch is contains a control message, 0 indicates that it doesn't. Control messages are used to enable transactions in Kafka and are generated by the broker. **Clients should not return control batches (ie. those with this bit set) to applications. (since 0.11.0.0)**


in https://github.com/dpkp/kafka-python/blob/master/kafka/consumer/fetcher.py#L458, the batch should be skipped if it's a control batch:
```python
            while batch is not None:
                if getattr(batch, 'is_control_batch', False):
                    continue
```

wdyt ?@dotnwat: We would be interested in helping add support at least for adding the ability to filter out control messages. It looks like this issue is a bit old, so I thought I'd check to see where this is at before jumping in.@jeffwidman: I don't think anyone has put together a PR, but if you did we'd be happy to review it.@nilansaha: Seems pretty old but any update ?@nilansaha: > https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol
> 
> > The sixth lowest bit indicates whether the RecordBatch includes a control message. 1 indicates that the RecordBatch is contains a control message, 0 indicates that it doesn't. Control messages are used to enable transactions in Kafka and are generated by the broker. **Clients should not return control batches (ie. those with this bit set) to applications. (since 0.11.0.0)**
> 
> in https://github.com/dpkp/kafka-python/blob/master/kafka/consumer/fetcher.py#L458, the batch should be skipped if it's a control batch:
> 
> ```python
>             while batch is not None:
>                 if getattr(batch, 'is_control_batch', False):
>                     continue
> ```
> 
> wdyt ?

Does this work ?@nilansaha: @dotnwat @jutley I have the same problem but I can't replicate this locally using kakfa zookeeper docker however. I can try making a PR if I can replicate it locally.@smalyshev: > in https://github.com/dpkp/kafka-python/blob/master/kafka/consumer/fetcher.py#L458, the batch should be skipped if it's a control batch:
> 
> ```python
>             while batch is not None:
>                 if getattr(batch, 'is_control_batch', False):
>                     continue
> ```
> 
> wdyt ?

This probably won't work, but if you add 
```
batch = records.next_batch()
```
before continue, then it might work. Also probably makes sense to put the condition here: https://github.com/dpkp/kafka-python/blob/master/kafka/consumer/fetcher.py#L467 so the offsets are correct?@bradenneal1: We recently struck the above issue. I've tested the proposed fix from @jouve / @nilansaha /@smalyshev and it works as intended. I've created a pull request under #2361

Solution patch:
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py
index b544e4b..9dd4b84 100644
--- a/kafka/consumer/fetcher.py
+++ b/kafka/consumer/fetcher.py
@@ -456,10 +456,20 @@ class Fetcher(six.Iterator):
             batch = records.next_batch()
             while batch is not None:
 
-                # LegacyRecordBatch cannot access either base_offset or last_offset_delta
+                # Try DefaultsRecordBatch / message log format v2
+                # base_offset, last_offset_delta, and control batches
                 try:
                     self._subscriptions.assignment[tp].last_offset_from_message_batch = batch.base_offset + \
                                                                                         batch.last_offset_delta
+                    # Control batches have a single record indicating whether a transaction
+                    # was aborted or committed.
+                    # When isolation_level is READ_COMMITTED (currently unsupported)
+                    # we should also skip all messages from aborted transactions
+                    # For now we only support READ_UNCOMMITTED and so we ignore the
+                    # abort/commit signal.
+                    if batch.is_control_batch:
+                        batch = records.next_batch()
+                        continue
                 except AttributeError:
                     pass
 
@@ -674,7 +684,7 @@ class Fetcher(six.Iterator):
                 if next_offset_from_batch_header > self._subscriptions.assignment[partition].position:
                     log.debug(
                         "Advance position for partition %s from %s to %s (last message batch location plus one)"
-                        " to correct for deleted compacted messages",
+                        " to correct for deleted compacted messages and/or transactional control records",
                         partition, self._subscriptions.assignment[partition].position, next_offset_from_batch_header)
                     self._subscriptions.assignment[partition].position = next_offset_from_batch_header
 
diff --git a/kafka/record/default_records.py b/kafka/record/default_records.py
index a098c42..b3a6fd0 100644
--- a/kafka/record/default_records.py
+++ b/kafka/record/default_records.py
@@ -269,8 +269,12 @@ class DefaultRecordBatch(DefaultRecordBase, ABCRecordBatch):
                 "payload, but instead read {}".format(length, pos - start_pos))
         self._pos = pos
 
-        return DefaultRecord(
-            offset, timestamp, self.timestamp_type, key, value, headers)
+        if self.is_control_batch:
+            return ControlRecord(
+                offset, timestamp, self.timestamp_type, key, value, headers)
+        else:
+            return DefaultRecord(
+                offset, timestamp, self.timestamp_type, key, value, headers)
 
     def __iter__(self):
         self._maybe_uncompress()
@@ -362,6 +366,45 @@ class DefaultRecord(ABCRecord):
         )
 
 
+class ControlRecord(DefaultRecord):
+    __slots__ = ("_offset", "_timestamp", "_timestamp_type", "_key", "_value",
+                 "_headers", "_version", "_type")
+
+    KEY_STRUCT = struct.Struct(
+        ">h"  # Current Version => Int16
+        "h"  # Type => Int16 (0 indicates an abort marker, 1 indicates a commit)
+    )
+
+    def __init__(self, offset, timestamp, timestamp_type, key, value, headers):
+        super(ControlRecord, self).__init__(offset, timestamp, timestamp_type, key, value, headers)
+        (self._version, self._type) = self.KEY_STRUCT.unpack(self._key)
+
+    # see https://kafka.apache.org/documentation/#controlbatch
+    @property
+    def version(self):
+        return self._version
+
+    @property
+    def type(self):
+        return self._type
+
+    @property
+    def abort(self):
+        return self._type == 0
+
+    @property
+    def commit(self):
+        return self._type == 1
+
+    def __repr__(self):
+        return (
+            "ControlRecord(offset={!r}, timestamp={!r}, timestamp_type={!r},"
+            " version={!r}, type={!r} <{!s}>)".format(
+                self._offset, self._timestamp, self._timestamp_type,
+                self._version, self._type, "abort" if self.abort else "commit")
+        )
+
+
 class DefaultRecordBatchBuilder(DefaultRecordBase, ABCRecordBatchBuilder):
 
     # excluding key, value and headers:


Test patch:
diff --git a/test/record/test_records.py b/test/record/test_records.py
index 5ed22d8..cab9592 100644
--- a/test/record/test_records.py
+++ b/test/record/test_records.py
@@ -60,6 +60,15 @@ record_batch_data_v0 = [
     b'\x00\xff\xff\xff\xff\x00\x00\x00\x03123'
 ]
 
+# Single record control batch (abort)
+control_batch_data_v2 = [
+    b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00R\x00\x00\x00\x00'
+    b'\x02e\x97\xff\xd0\x00\x20\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    b'\x98\x96\x7f\x00\x00\x00\x00\x00\x98\x96'
+    b'\x7f\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff'
+    b'\x00\x00\x00\x01@\x00\x00\x00\x08\x00\x00\x00\x00,opaque-control-message\x00'
+]
+
 
 def test_memory_records_v2():
     data_bytes = b"".join(record_batch_data_v2) + b"\x00" * 4
@@ -230,3 +239,18 @@ def test_memory_records_builder_full(magic, compression_type):
         key=None, timestamp=None, value=b"M")
     assert metadata is None
     assert builder.next_offset() == 1
+
+
+def test_control_record_v2():
+    data_bytes = b"".join(control_batch_data_v2)
+    records = MemoryRecords(data_bytes)
+
+    assert records.has_next() is True
+    batch = records.next_batch()
+    assert batch.is_control_batch is True
+    recs = list(batch)
+    assert len(recs) == 1
+    assert recs[0].version == 0
+    assert recs[0].type == 0
+    assert recs[0].abort is True
+    assert recs[0].commit is False