Script 'mail_helper' called by obssrc
Hello community,
here is the log from the commit of package python-kafka-python for
openSUSE:Factory checked in at 2026-07-07 21:03:21
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Comparing /work/SRC/openSUSE:Factory/python-kafka-python (Old)
and /work/SRC/openSUSE:Factory/.python-kafka-python.new.1982 (New)
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Package is "python-kafka-python"
Tue Jul 7 21:03:21 2026 rev:15 rq:1364124 version:3.0.7
Changes:
--------
--- /work/SRC/openSUSE:Factory/python-kafka-python/python-kafka-python.changes
2026-06-28 21:11:55.673001471 +0200
+++
/work/SRC/openSUSE:Factory/.python-kafka-python.new.1982/python-kafka-python.changes
2026-07-07 21:05:51.103758828 +0200
@@ -1,0 +2,7 @@
+Mon Jul 6 20:40:25 UTC 2026 - Dirk Müller <[email protected]>
+
+- update to 3.0.7:
+ * transport: cancel pending io coroutines on close()
+ * producer: fix stalled shutdown when brokers unreachable
+
+-------------------------------------------------------------------
Old:
----
python-kafka-python-3.0.6.tar.gz
New:
----
python-kafka-python-3.0.7.tar.gz
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Other differences:
------------------
++++++ python-kafka-python.spec ++++++
--- /var/tmp/diff_new_pack.OMEEMG/_old 2026-07-07 21:05:52.071792713 +0200
+++ /var/tmp/diff_new_pack.OMEEMG/_new 2026-07-07 21:05:52.075792854 +0200
@@ -23,7 +23,7 @@
%endif
Name: python-kafka-python
-Version: 3.0.6
+Version: 3.0.7
Release: 0
Summary: Pure Python client for Apache Kafka
License: Apache-2.0
++++++ python-kafka-python-3.0.6.tar.gz -> python-kafka-python-3.0.7.tar.gz
++++++
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/kafka-python-3.0.6/CHANGES.md
new/kafka-python-3.0.7/CHANGES.md
--- old/kafka-python-3.0.6/CHANGES.md 2026-06-25 21:02:57.000000000 +0200
+++ new/kafka-python-3.0.7/CHANGES.md 2026-06-29 00:11:16.000000000 +0200
@@ -1,9 +1,14 @@
+# 3.0.7 (Jun 28, 2026)
+
+### Fixes
+* transport: cancel pending io coroutines on close() (#3109)
+* producer: fix stalled shutdown when brokers unreachable (#3108)
+
# 3.0.6 (Jun 25, 2026)
### Fixes
* consumer: track current leader epoch in addition to record epoch (#3107)
-
### CI
* Also `wait_for_topics` on <0.10 test fixtures (#3105)
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/kafka-python-3.0.6/docs/changelog.rst
new/kafka-python-3.0.7/docs/changelog.rst
--- old/kafka-python-3.0.6/docs/changelog.rst 2026-06-25 21:02:57.000000000
+0200
+++ new/kafka-python-3.0.7/docs/changelog.rst 2026-06-29 00:11:16.000000000
+0200
@@ -1,6 +1,15 @@
Changelog
=========
+3.0.7 (Jun 28, 2026)
+####################
+
+Fixes
+-----
+* transport: cancel pending io coroutines on close() (#3109)
+* producer: fix stalled shutdown when brokers unreachable (#3108)
+
+
3.0.6 (Jun 25, 2026)
####################
@@ -8,7 +17,6 @@
-----
* consumer: track current leader epoch in addition to record epoch (#3107)
-
CI
--
* Also `wait_for_topics` on <0.10 test fixtures (#3105)
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/kafka-python-3.0.6/kafka/net/transport.py
new/kafka-python-3.0.7/kafka/net/transport.py
--- old/kafka-python-3.0.6/kafka/net/transport.py 2026-06-25
21:02:57.000000000 +0200
+++ new/kafka-python-3.0.7/kafka/net/transport.py 2026-06-29
00:11:16.000000000 +0200
@@ -19,6 +19,8 @@
self._closed = False
self._write_buffer = deque()
self._writing = False
+ self._read_task = None
+ self._write_task = None
self._protocol = None
self._read = False
self._write = True
@@ -80,7 +82,7 @@
data_received() method.
"""
if not self._read:
- self._net.call_soon(self._read_from_sock)
+ self._read_task = self._net.call_soon(self._read_from_sock)
self._read = True
log.debug('%s: Resumed reading', self)
@@ -173,7 +175,7 @@
self._write_buffer.append(data)
if not self._writing:
self._writing = True
- self._net.call_soon(self._write_to_sock)
+ self._write_task = self._net.call_soon(self._write_to_sock)
def writelines(self, list_of_data):
"""Write a list (or any iterable) of data bytes to the transport."""
@@ -182,7 +184,7 @@
self._write_buffer.extend(list_of_data)
if not self._writing:
self._writing = True
- self._net.call_soon(self._write_to_sock)
+ self._write_task = self._net.call_soon(self._write_to_sock)
async def _write_to_sock(self):
try:
@@ -272,6 +274,10 @@
except OSError:
pass
sock.close()
+ for task in (self._read_task, self._write_task):
+ if task is not None:
+ self._net.cancel(task)
+ self._read_task = self._write_task = None
proto = self._protocol
self._protocol = None
if proto is not None:
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/kafka-python-3.0.6/kafka/producer/kafka.py
new/kafka-python-3.0.7/kafka/producer/kafka.py
--- old/kafka-python-3.0.6/kafka/producer/kafka.py 2026-06-25
21:02:57.000000000 +0200
+++ new/kafka-python-3.0.7/kafka/producer/kafka.py 2026-06-29
00:11:16.000000000 +0200
@@ -646,7 +646,7 @@
_self = weakref.proxy(self)
def wrapper():
try:
- _self.close(timeout=0, null_logger=True)
+ _self.close(timeout=1, null_logger=True)
except (ReferenceError, AttributeError):
pass
return wrapper
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/kafka-python-3.0.6/kafka/producer/sender.py
new/kafka-python-3.0.7/kafka/producer/sender.py
--- old/kafka-python-3.0.6/kafka/producer/sender.py 2026-06-25
21:02:57.000000000 +0200
+++ new/kafka-python-3.0.7/kafka/producer/sender.py 2026-06-29
00:11:16.000000000 +0200
@@ -310,7 +310,7 @@
return False
log.debug("%s: Sending transactional request %s", str(self),
next_request_handler.request)
- while not self._force_close:
+ while self._running and not self._force_close:
target_node = None
try:
if next_request_handler.needs_coordinator():
@@ -325,7 +325,8 @@
else:
target_node = self._client.least_loaded_node()
if target_node is None:
-
self._client.poll(future=self._metadata.request_update())
+
self._client.poll(timeout_ms=self.config['retry_backoff_ms'],
+
future=self._metadata.request_update())
elif not self._client.await_ready(target_node,
timeout_ms=self.config['request_timeout_ms']):
continue
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/kafka-python-3.0.6/kafka/version.py
new/kafka-python-3.0.7/kafka/version.py
--- old/kafka-python-3.0.6/kafka/version.py 2026-06-25 21:02:57.000000000
+0200
+++ new/kafka-python-3.0.7/kafka/version.py 2026-06-29 00:11:16.000000000
+0200
@@ -1 +1 @@
-__version__ = '3.0.6'
+__version__ = '3.0.7'
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/kafka-python-3.0.6/test/net/test_transport.py
new/kafka-python-3.0.7/test/net/test_transport.py
--- old/kafka-python-3.0.6/test/net/test_transport.py 2026-06-25
21:02:57.000000000 +0200
+++ new/kafka-python-3.0.7/test/net/test_transport.py 2026-06-29
00:11:16.000000000 +0200
@@ -6,7 +6,7 @@
import kafka.errors as Errors
from kafka.future import Future
-from kafka.net.selector import NetworkSelector
+from kafka.net.selector import NetworkSelector, TaskState
from kafka.net.transport import KafkaTCPTransport
@@ -365,3 +365,58 @@
t._closed = True
s = str(t)
assert 'closed' in s
+
+
+class TestTransportWaiterCleanup:
+ """Regression: a locally-initiated close()/abort() must reclaim the socket
+ read/write coroutine tasks parked in the event loop.
+
+ These tests fail until ``KafkaTCPTransport._close`` cancels its read/write
+ waiter tasks (``net.cancel(task)``); the selector's existing WAIT_IO branch
+ in ``cancel()`` then drives the io_guard finalizer and discards the task.
+ """
+
+ def test_local_close_reclaims_parked_reader(self, net, socketpair):
+ rsock, wsock = socketpair
+ t = KafkaTCPTransport(net, wsock)
+ t.set_protocol(MagicMock())
+ baseline = len(net._pending_tasks)
+
+ t.resume_reading()
+ net.drain() # reader runs and parks in WAIT_IO on wait_read(wsock)
+ parked = [task for task in net._pending_tasks if task.state is
TaskState.WAIT_IO]
+ assert len(parked) == 1, 'reader did not park as expected'
+ assert len(net._pending_tasks) == baseline + 1
+
+ # Empty write buffer -> close() tears the socket down synchronously.
The
+ # peer (rsock) never sent anything, so there is no I/O event to wake
the
+ # parked reader; close() itself must reclaim it.
+ t.close()
+ net.drain() # running the loop must not be needed -- and must not
help either
+
+ assert t._sock is None
+ assert len(net._pending_tasks) == baseline, (
+ 'parked reader leaked into _pending_tasks after local close')
+ assert not any(task.state is TaskState.WAIT_IO for task in
net._pending_tasks)
+
+ def test_local_close_reclaims_parked_writer(self, net, socketpair):
+ rsock, wsock = socketpair
+ t = KafkaTCPTransport(net, wsock)
+ t.set_protocol(MagicMock())
+ baseline = len(net._pending_tasks)
+
+ # write() schedules _write_to_sock, whose loop parks on the first
+ # wait_write before the bytes leave the buffer. drain() steps it once
+ # to the park and returns (no _ready left), leaving it suspended.
+ t.write(b'data')
+ net.drain()
+ parked = [task for task in net._pending_tasks if task.state is
TaskState.WAIT_IO]
+ assert len(parked) == 1, 'writer did not park as expected'
+
+ t.abort(error=Errors.KafkaConnectionError('boom'))
+ net.drain()
+
+ assert t._sock is None
+ assert len(net._pending_tasks) == baseline, (
+ 'parked writer leaked into _pending_tasks after abort')
+ assert not any(task.state is TaskState.WAIT_IO for task in
net._pending_tasks)
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/kafka-python-3.0.6/test/producer/test_kip679_config.py
new/kafka-python-3.0.7/test/producer/test_kip679_config.py
--- old/kafka-python-3.0.6/test/producer/test_kip679_config.py 2026-06-25
21:02:57.000000000 +0200
+++ new/kafka-python-3.0.7/test/producer/test_kip679_config.py 2026-06-29
00:11:16.000000000 +0200
@@ -12,66 +12,52 @@
# same process. Observable behavior on the producer instance
(enable_idempotence
# flag, _transaction_manager presence, config['acks']) is what callers actually
# depend on, and is what these tests verify.
+#
+# These producers point at the default localhost:9092 with no broker running,
so
+# the `with KafkaProducer(...) as p` form relies on close() tearing down the
+# Sender thread even though it never connects (see
Sender._maybe_send_pending_request).
class TestDefaultsAreIdempotent:
"""KIP-679: defaults instantiate an idempotent producer with acks=all."""
def test_bare_construct_is_idempotent(self):
- p = KafkaProducer(api_version=(0, 11))
- try:
+ with KafkaProducer(api_version=(0, 11)) as p:
assert p.config['enable_idempotence'] is True
assert p.config['acks'] == -1
assert p._transaction_manager is not None
assert p._transaction_manager.is_transactional() is False
- finally:
- p.close(timeout=0)
def test_opt_out_disables_idempotence_but_keeps_acks_default(self):
- p = KafkaProducer(enable_idempotence=False, api_version=(0, 11))
- try:
+ with KafkaProducer(enable_idempotence=False, api_version=(0, 11)) as p:
assert p.config['enable_idempotence'] is False
assert p.config['acks'] == -1
assert p._transaction_manager is None
- finally:
- p.close(timeout=0)
class TestDefaultIdempotenceSilentDisable:
"""User-provided incompatible config + default idempotence -> silent
disable."""
def test_acks_1_silently_disables(self):
- p = KafkaProducer(acks=1, api_version=(0, 11))
- try:
+ with KafkaProducer(acks=1, api_version=(0, 11)) as p:
assert p.config['enable_idempotence'] is False
assert p.config['acks'] == 1
assert p._transaction_manager is None
- finally:
- p.close(timeout=0)
def test_acks_0_silently_disables(self):
- p = KafkaProducer(acks=0, api_version=(0, 11))
- try:
+ with KafkaProducer(acks=0, api_version=(0, 11)) as p:
assert p.config['enable_idempotence'] is False
assert p._transaction_manager is None
- finally:
- p.close(timeout=0)
def test_retries_zero_silently_disables(self):
- p = KafkaProducer(retries=0, api_version=(0, 11))
- try:
+ with KafkaProducer(retries=0, api_version=(0, 11)) as p:
assert p.config['enable_idempotence'] is False
assert p._transaction_manager is None
- finally:
- p.close(timeout=0)
def test_max_in_flight_too_high_silently_disables(self):
- p = KafkaProducer(max_in_flight_requests_per_connection=10,
api_version=(0, 11))
- try:
+ with KafkaProducer(max_in_flight_requests_per_connection=10,
api_version=(0, 11)) as p:
assert p.config['enable_idempotence'] is False
assert p._transaction_manager is None
- finally:
- p.close(timeout=0)
class TestOldBrokerFallback:
@@ -79,14 +65,11 @@
acks=-1 retained (acks=all is valid on any wire version)."""
def test_old_broker_silently_disables(self):
- p = KafkaProducer(api_version=(0, 10))
- try:
+ with KafkaProducer(api_version=(0, 10)) as p:
assert p.config['enable_idempotence'] is False
# Per KIP-679 + project decision: acks stays at -1 even on old
brokers.
assert p.config['acks'] == -1
assert p._transaction_manager is None
- finally:
- p.close(timeout=0)
class TestExplicitIdempotenceConflicts:
@@ -139,16 +122,13 @@
def test_acks_all_string_accepted(self):
"""'all' normalizes to -1 before the conflict check fires."""
- p = KafkaProducer(
+ with KafkaProducer(
enable_idempotence=True,
acks='all',
api_version=(0, 11),
- )
- try:
+ ) as p:
assert p.config['acks'] == -1
assert p._transaction_manager is not None
- finally:
- p.close(timeout=0)
def test_old_broker_with_explicit_idempotence_raises(self):
with pytest.raises(Errors.KafkaConfigurationError,
match="api_version"):
