Fix DROP TABLE hang on Kafka tables after consumer heartbeat error

[alexey-milovidov]

When a Kafka consumer experiences a heartbeat error followed by a rebalance, DROP TABLE can hang indefinitely. The shutdown sequence waits for the streaming task to finish, but the task is stuck inside CompletedPipelineExecutor::execute because consumers are blocked in librdkafka's poll_batch during the rebalance — and the pipeline has no external cancellation mechanism.

Use CompletedPipelineExecutor::setCancelCallback to allow the pipeline to be cancelled promptly when shutdown_called is set. The callback checks every 100 ms and triggers pipeline cancellation, letting streamToViews return quickly so the shutdown sequence can proceed.

Skip offset commits after cancellation: the cancelled pipeline may not have fully written data to dependent views, so committing offsets would cause data loss. The consumer is marked as dirty instead, and offsets remain uncommitted.

CI report: https://s3.amazonaws.com/clickhouse-test-reports/json.html?REF=master&sha=933424a0835a809944686507ee5c8a40f5d440e4&name_0=MasterCI&name_1=Integration%20tests%20%28amd_asan_ubsan%2C%20db%20disk%2C%20old%20analyzer%2C%202%2F6%29

Changelog category (leave one):

• Bug Fix (user-visible misbehavior in an official stable release)

Changelog entry (a user-readable short description of the changes that goes into CHANGELOG.md):

Fix DROP TABLE hanging indefinitely on Kafka engine tables when consumers are stuck in a rebalance after a heartbeat error.

Documentation entry for user-facing changes

• Documentation is written (mandatory for new features)

[clickhouse-gh]

Workflow [PR], commit [10112ad]

Summary: ✅

---

AI Review

Summary

This PR addresses DROP TABLE hangs for Kafka tables by adding pipeline cancellation and skipping offset commits during shutdown. The fix direction is good, but there is one blocker: the new shutdown guard around source->commit() has a race and can still commit offsets after shutdown begins, which can reintroduce data-loss behavior the patch intends to prevent.

Missing context

• ⚠️ No CI logs or test artifacts were reviewed beyond PR description, so runtime validation of the new shutdown/cancellation path is not confirmed.

Findings

• ❌ Blockers
  • [src/Storages/Kafka/StorageKafka.cpp:738] TOCTOU race in if (!shutdown_called) source->commit();.
    shutdown_called can flip to true immediately after the check, and commit may still execute after cancellation starts. That can commit offsets for data not fully flushed to dependent views.
    Suggested fix: track cancellation outcome for this execution (e.g., local cancelled_by_shutdown flag set by cancel callback) and skip commit based on that state instead of a late atomic read.

Tests

• ⚠️ Add a regression test that starts Kafka ingestion with attached MVs, triggers shutdown/cancellation while processing, and verifies offsets are not committed in the cancelled batch.

ClickHouse Rules

Item	Status	Notes
Deletion logging	➖
Serialization versioning	➖
Core-area scrutiny	⚠️	Core streaming/shutdown path changed; one race remains in commit gating during cancellation.
No test removal	✅
Experimental gate	➖
No magic constants	✅
Backward compatibility	⚠️	Potential offset commit after shutdown can cause message loss semantics for existing users.
SettingsChangesHistory.cpp	➖
PR metadata quality	✅
Safe rollout	⚠️	Rollout is unsafe until commit/cancellation race is fixed.
Compilation time	✅

Performance & Safety

• Cancellation is implemented via CompletedPipelineExecutor::setCancelCallback, but safety depends on correctly suppressing commit for cancelled execution. The current TOCTOU around shutdown_called leaves a data-loss window.

Final Verdict

• Status: ⚠️ Request changes
• Minimum required actions:
  • Fix commit gating so a shutdown-triggered cancellation cannot race into source->commit().
  • Add a regression test covering shutdown cancellation vs offset commit behavior.

[tuanpach]

I think we should print some logs when shutdown_called is true.

[tuanpach]

This is a minor comment.

[clickhouse-gh]

❌ There is a TOCTOU race here: if (!shutdown_called) can still call source->commit() if shutdown_called flips to true right after the check.

That re-introduces the data-loss case this patch is trying to avoid (commit offsets even though shutdown cancellation started and the pipeline may not have fully flushed to MVs).

Please gate commit on a local cancelled_by_shutdown flag set from the cancel callback (or another state that reflects whether this execute was actually cancelled), instead of a late read of shutdown_called.

[clickhouse-gh]

LLVM Coverage Report

Metric	Baseline	Current	Δ
Lines	84.10%	84.00%	-0.10%
Functions	24.50%	24.70%	+0.20%
Branches	76.60%	76.60%	+0.00%

PR changed lines: PR changed-lines coverage: 100.00% (16/16, 0 noise lines excluded)
Diff coverage report
Uncovered code

[azat]

While working on #101217

I realized that this may not be a good enough fix:

• first it may create duplicates, due to commit was not called
• secondly, this check is racy

Maybe it is better to revert this and fix kafka differently - since after this change the tests fails more often, @alexey-milovidov @tuanpach @antaljanosbenjamin WDYT?

Also, maybe this will help with fixing DROP - #100612

I will prepare a revert for now -#101646