Skip to content

Add a Consumer::commit overload bounded by a timeout - #8

Open
kalavt wants to merge 2 commits into
ClickHouse:ClickHouse/release-0.4.1from
kalavt:feature/commit-with-timeout
Open

Add a Consumer::commit overload bounded by a timeout#8
kalavt wants to merge 2 commits into
ClickHouse:ClickHouse/release-0.4.1from
kalavt:feature/commit-with-timeout

Conversation

@kalavt

@kalavt kalavt commented Aug 10, 2026

Copy link
Copy Markdown

Requested in ClickHouse/ClickHouse#113921, where the same timeout was first written on the ClickHouse side against librdkafka's C API: "Can we patch cppkafka instead? This is the perfect fit for it and we can also upstream the changes."

Consumer::commit calls rd_kafka_commit(handle, offsets, 0), which takes no timeout and waits on librdkafka's reply queue until an op arrives. When no broker is reachable at the moment of the commit, librdkafka enqueues no reply and the call does not return. offsets.commit.timeout.ms does not bound this - it bounds a broker's response to an OffsetCommit request, not the case where there is no reachable broker to send one to. In ClickHouse this stops one Kafka table permanently: StorageKafka::threadFunc never returns, so it never reaches the scheduleAfter at its exit and the consumer is never returned to the pool. Measured on a 26.7.2.59 server, one table frozen 9h33m, cleared only by a pod restart:

pthread_cond_wait
cnd_timedwait_abs
rd_kafka_q_pop_serve0
rd_kafka_commit
DB::KafkaConsumer::commit()
DB::StorageKafka::threadFunc(unsigned long)
DB::BackgroundSchedulePool::threadFunction()

The new overloads commit through rd_kafka_commit_queue on a queue private to the call and wait on it with Queue::next_event(timeout), throwing HandleException carrying RD_KAFKA_RESP_ERR__TIMED_OUT once the deadline passes. Dropping the queue discards a late result, which is safe: an offset commit is idempotent.

  • commit(std::chrono::milliseconds) and commit(const TopicPartitionList&, std::chrono::milliseconds), following the shape get_offsets_committed(list, timeout) already uses rather than adding a mode flag.
  • Additive: no data members, no virtual functions, and the existing overloads are untouched. std::chrono::duration's constructor from a representation is explicit, so no existing call site becomes ambiguous. Every translation unit under src/ compiles against the changed header.
  • Event::get_partitions_error comes with it: an RD_KAFKA_EVENT_OFFSET_COMMIT reply carries per-partition errors separately from its request-level one, so without it the timeout overloads would report success on a commit the broker rejected for one partition, where the blocking overloads throw through check_error(error, list).

tests/commit_timeout_test.cpp covers both overloads and needs no KAFKA_TEST_INSTANCE: nothing accepts connections on port 1, so the group never gets a coordinator and librdkafka enqueues no commit reply, which is the case the timeout exists for. Run it alone with cppkafka_tests "[commit_timeout]". Removing the stored offset from the fixture turns the case red on the error code alone while the elapsed-time bound still passes - a commit carrying no offset is answered _NO_OFFSET in 0ms, so a case asserting only "it returned in time" would not have caught it. This repository runs no CI, so both sections were run locally against librdkafka 2.14.1; the same behaviour is covered from the ClickHouse side in ClickHouse/ClickHouse#113921.

Same patch opened against upstream, where it applies cleanly: mfontanini#325

For whoever does the eventual contrib/cppkafka bump in ClickHouse: master and 26.7 both pin 8cc2f31, which is 3 commits behind this branch, so the bump also brings #7 (the handle_rebalance exception changes) with it.

`Consumer::commit` calls `rd_kafka_commit`, which takes no timeout and waits on its
reply queue until an op arrives. When every broker is unreachable at the moment of
the commit, librdkafka enqueues no reply and the call does not return, so a caller
has no way to bound it. `offsets.commit.timeout.ms` bounds a broker's response to an
OffsetCommit request, not the case where no broker is reachable to send one to.

The new overloads commit through `rd_kafka_commit_queue` on a queue private to the
call and wait on it with `Queue::next_event(timeout)`, throwing `HandleException`
carrying `RD_KAFKA_RESP_ERR__TIMED_OUT` once the deadline passes. Dropping the queue
discards a late result, which is safe: an offset commit is idempotent.

`Event::get_partitions_error` is added alongside, because the commit reply carries
per-partition errors separately from its request-level one. Without it the timeout
overloads would report success on a commit the broker rejected for one partition,
while the blocking overloads throw through `check_error(error, list)`.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
@kalavt
kalavt force-pushed the feature/commit-with-timeout branch from 0af9c68 to fab890e Compare August 10, 2026 14:43
kalavt added a commit to kalavt/ClickHouse that referenced this pull request Aug 10, 2026
`TopicPartitionsListPtr`'s deleter is a function pointer, so the type has no default
constructor and the previous declaration did not compile. `make_handle` builds the
empty one, which `rd_kafka_commit_queue` reads as "commit the current assignment".

Two more, both from review:

- `KafkaConsumer::rewindToLastCommitted`, reached from `KafkaSource::generateImpl`,
  still committed through the unbounded path, so the abort case could wedge the
  streaming task the same way. It goes through the deadline now.
- the commit reply carries per-partition errors separately from its request-level
  error, and cppkafka's blocking commits throw on them through
  `check_error(error, list)`. The bounded path now does too.

The helper stays for now. It moves into cppkafka through ClickHouse/cppkafka#8 once
that lands; keeping it here is what lets this branch compile and produce a build to
verify against a live cluster in the meantime.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The case needs no KAFKA_TEST_INSTANCE: nothing accepts connections on port 1, so the group never gets
a coordinator and librdkafka enqueues no commit reply. Removing the stored offset from the fixture
turns it red on the error code alone - the elapsed-time bound still passes, since a commit carrying no
offset is answered `_NO_OFFSET` in 0ms.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant