confluentinc / confluentinc/confluent-kafka-python
on_revoke callback passed partition list with partition not already in assignment after consumer group session timeout
- Dominant language
- Python
- Stars
- 509
- Forks
- 964
- Avg merge
- 2d 2h
- Merged PRs (30d)
- 14
Description
Description
===========
We are using a single confluent_kafka.Consumer for a partition group that is always unique (via generating a fresh uuid4 in the group)
We use on_assign and on_revoke callbacks in the consumer with the consumer.subscribe call
Our on_assign callback looks like this:
```
def _set_and_assign_offsets(self, consumer: ConfluentConsumer, partitions: List[TopicPartition]):
self._set_offsets_by_script_state(consumer, partitions)
self._event_loop.call_soon_threadsafe(self._assignment_sync_event.set)
def _set_offsets_by_script_state(self, consumer: ConfluentConsumer, partitions: List[TopicPartition]):
for p in partitions:
if p.topic in self._begin_topics or p.topic in self._snapshot_topics:
p.offset = OFFSET_BEGINNING if self._initialising else OFFSET_STORED
elif p.topic in self._end_topics:
p.offset = OFFSET_END if self._initialising else OFFSET_STORED
consumer.incremental_assign(partitions)
```
Our on_revoke callback looks like this:
def _on_revoke(self, consumer: ConfluentConsumer, partitions: List[TopicPartition]):
_logger.info(f'These partitions have been revoked from the consumer {partitions}')
consumer.incremental_unassign(partitions)
At some point we also use consumer.incremental_unassign to pause consumption for self._snapshot_topics
We use batch consume to consume messages from this consumer (num_messages > 1)
I think when our broker is doing some daily cleanup, we get kicked from the cluster leader, but it seems that the on_revoke callback is passed topic partitions that the consumer no longer has an assignment to
There is only 1 consumer in that cg, so we would expect all of TOPIC_2 to be assigned to this consumer
Also, this seems to happen without the on_revoke callback also. Initially we were happy with the defaults
```
"2023-04-10T13:05:12.080Z","","%4|1681131912.080|SESSTMOUT|| [thrd:main]: Consumer group session timed out (in join-state steady) after 10498 ms without a successful response from the group coordinator (broker 3, last error was Success): revoking assignment and rejoining group"
"2023-04-10T13:05:12.081Z","","%4|1681131912.081|ASSIGN|| [thrd:main]: Group """": application *assign() call failed: [0] can't be unassigned since it is not in the current assignment"
"2023-04-10T13:05:12.081Z","","event='These partitions have been revoked from the consumer [TopicPartition{topic=,partition=0,offset=-1001,error=None}, TopicPartition{topic=,partition=1,offset=-1001,error=None}, TopicPartition{topic=,partition=2,offset=-1001,error=None}, TopicPartition{topic=,partition=3,offset=-1001,error=None}, TopicPartition{topic=,partition=4,offset=-1001,error=None}, TopicPartition{topic=,partition=0,offset=-1001,error=None}, TopicPartition{topic=,partition=1,offset=-1001,error=None}, TopicPartition{topic=,partition=2,offset=-1001,error=None}, TopicPartition{topic=,partition=3,offset=-1001,error=None}, TopicPartition{topic=,partition=4,offset=-1001,error=None}, TopicPartition{topic=,partition=0,offset=-1001,error=None}, TopicPartition{topic=,partition=1,offset=-1001,error=None}, TopicPartition{topic=,partition=2,offset=-1001,error=None}, TopicPartition{topic=,partition=3,offset=-1001,error=None}, TopicPartition{topic=,partition=4,offset=-1001,error=None}]' logger='.kafka.Consumer' level='info' timestamp='2023-04-10T13:05:12.080859Z'"
```
How to reproduce
================
Using confluent_kafka 2.0.2 and librdkafka 2.0.2
Broker version is 2.6-IV0
Use the above setup and then maybe shut down the leader to guarantee a repro. We get this nightly so it is pretty consistent
Checklist
=========
Please provide the following information:
- [x] confluent-kafka-python and librdkafka version (`confluent_kafka.version()` and `confluent_kafka.libversion()`):
- [x] Apache Kafka broker version: 2.6-IV0
- [x] Client configuration: `{'auto.offset.reset': 'smallest',
'partition.assignment.strategy': 'cooperative-sticky',
'check.crcs': True,
'connections.max.idle.ms': 540000,
'enable.auto.commit': True,
'enable.partition.eof': True,
'fetch.wait.max.ms': 20,
'max.in.flight': 5,
'metadata.max.age.ms': 300000,
'reconnect.backoff.max.ms': 1000,
'reconnect.backoff.ms': 50,
'session.timeout.ms': 10000,
'socket.receive.buffer.bytes': 16777216,
'socket.send.buffer.bytes': 16777216,
'socket.timeout.ms': 30000}`
- [x] Operating system: CentOS 8 on Docker
- [x] Provide client logs (with `'debug': '..'` as necessary)
- [ ] Provide broker log excerpts
- [x] Critical issue: No - working around by restarting the application
Contributor guide
No contributing guide indexed for this repository
Research direction
No repository file or test is named. Start by tracing the on_assign/on_revoke callbacks and incremental_assign/incremental_unassign behavior, then reproduce with confluent_kafka 2.0.2, librdkafka 2.0.2, and a broker or group-coordinator interruption. Done means the session-timeout path no longer attempts to unassign partitions absent from the current assignment, with the callback behavior covered by a regression test.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 42/100