[VibeSWE]

Problem

The Kafka connector took the destination tenant from producer-controlled message headers. Any producer able to write to a consumed topic could inject data into another tenant.

Fix

Resolve the destination tenant only from --consumer-tenant-id / P_KAFKA_CONSUMER_TENANT_ID, using the same configured destination for every consumed topic. Headers can assert that destination but cannot select another tenant.

• Multi-tenancy off: tenant headers are ignored and records use the default destination.
• Multi-tenancy on with a tenant configured: headerless records and matching headers use the configured tenant. Mismatched records, conflicting duplicate headers, null headers, and invalid tenant header values are dropped and counted in parseable_kafka_tenant_mismatch_dropped_total.
• Multi-tenancy on with no tenant configured: headerless records go to the default tenant; all records carrying an exact x-p-tenant header are dropped and counted, including headers whose value is DEFAULT_TENANT (the default tenant's literal ID). A startup warning recommends configuring the destination.
• Startup validation rejects a configured tenant unless multi-tenancy is enabled and the value passes the server's tenant-ID validator. An explicitly empty option or environment value is also rejected by the nonempty value parser. Leave P_KAFKA_CONSUMER_TENANT_ID unset, not empty.
• A shared header-extraction helper is exercised with real Kafka OwnedHeaders, including exact key matching and null values.

Rejected records now advance the committed offset even when no later accepted record arrives. ConsumerRecord::from_message returns an offset-only marker for a rejected record: payload and key are None, the tenant remains the trusted configured destination, and topic, partition, offset, and timestamp are preserved. Markers pass through the ordered partition queue alongside accepted records. All-marker chunks (and tombstone-only chunks) skip prepare_stream and commit their next offset without accessing stream metadata or staging data.

Within each partition, ordered buffering allows up to MAX_IN_FLIGHT_CHUNKS_PER_PARTITION = 8 chunks in flight. Chunk processing can complete out of order, but commit offsets are yielded in input order. A fast marker-only tail therefore cannot commit past an earlier unfinished chunk. Existing commit-after-processing-error behavior is preserved; an ordinary error completes that chunk under the existing drop policy.

The generic ConsumerRecord::from_message<M: Message> and KafkaStreams::handle_message<M: Message> seams use the same production conversion and routing logic with borrowed broker messages and owned messages in tests. Tests can verify real header extraction, marker metadata, and delivery to the partition receiver without a Kafka broker.

Markers and genuine Kafka tombstones are handled identically: neither emits the former "without a payload" warning, and a chunk containing only tombstones or markers no longer creates a stream or accesses its metadata.

Mismatch warnings are rate-limited to at most one every 60 seconds per process. Each warning describes rejected x-p-tenant headers neutrally and includes the triggering record's topic, partition, tenant_configured boolean, and interval drop count across all partitions. Header values, keys, and payloads are never logged. Warnings repeat only when another drop occurs after the interval; the Prometheus counter remains unlabeled.

Upgrade note

Multi-tenant users who relied on message headers to choose a destination must run an ingestor configured with --consumer-tenant-id / P_KAFKA_CONSUMER_TENANT_ID for each destination tenant. Each tenant needs its own consumer group or its own topics. When consuming shared topics, assign a distinct --consumer-group-id / P_KAFKA_CONSUMER_GROUP_ID to each tenant's ingestors; do not leave different tenants on the shared default parseable-connect-cg. A shared group distributes partitions among tenants' ingestors, which can drop records intended for another tenant and commit past them, causing data loss. Ingestors for the same tenant may share that tenant's group.

Leaving the tenant option unset in multi-tenant mode accepts only headerless records at the default destination. Any record with an exact x-p-tenant header is dropped, including one naming DEFAULT_TENANT. The reserved default tenant ID cannot be configured explicitly. Single-tenant mode continues to ignore tenant headers.

Configure the destination tenant and tenant-specific consumer groups before upgrading. Rejected records are now committed, so a misconfigured rollout permanently skips them for normal consumption, even after configuration is corrected. Recovery requires a manual consumer-group offset reset with kafka-consumer-groups --reset-offsets. Leave P_KAFKA_CONSUMER_TENANT_ID unset, not empty, when using the unconfigured behavior; an empty value fails startup parsing.

Tests

• rejected_message_becomes_marker_with_trusted_tenant_and_broker_metadata: real OwnedMessage conversion strips rejected payload/key, preserves broker metadata and configured tenant, and computes committed next offset 43 from source offset 42.
• accepted_message_preserves_payload_and_key: accepted OwnedMessage conversion preserves content and broker metadata.
• rejected_record_is_delivered_to_partition_stream_receiver: exercises the real routing path and receives the rejected record's marker from its partition receiver. A temporary early-return mutation made this test fail (0 passed, 1 failed, exit 101: no partition receiver); the mutation was restored before validation.
• all_rejected_partition_commits_next_offset: marker-only chunks produce ordered committed next offsets 12 and 13 using a fake processor/commit seam.
• chunk_needs_stream_only_for_payload_records: pure predicate checks empty, marker/tombstone-only, and mixed chunks without global PARSEABLE initialization.
• rejected_tail_waits_for_accepted_records_before_commit: gates accepted processing and proves a rejected tail cannot commit past it.
• later_chunk_starts_while_earlier_chunk_blocks_commits: later processing starts while the first chunk is blocked; both commits wait and then yield in source order.
• tenant_drop_warnings_repeat_after_sixty_seconds_with_interval_count: verifies monotonic warning intervals and accumulated drop counts.
• owned_unrelated_headers_do_not_hide_tenant_mismatch: real header extraction retains mismatches among unrelated headers.
• owned_case_variant_key_does_not_select_tenant: header keys match exactly.
• owned_matching_tenant_header_is_accepted: matching headers pass without a drop.
• owned_null_tenant_header_is_dropped: null values are preserved for rejection.
• absent_headers_preserve_configured_or_default_destination: absent headers retain trusted routing.
• empty_consumer_tenant_id_is_rejected_by_cli: empty CLI option fails parsing.
• empty_consumer_tenant_env_is_rejected_in_child_process / tenant_env_child: subprocess isolation verifies empty environment-value rejection without global environment mutation or test races. The child is ignored in normal runs; its parent explicitly invokes it with --ignored --exact, sets the guard and empty tenant environment value, and checks that one test passed.

Restart coverage is unit-level: these tests verify the committed next offset, not live-broker restart/reassignment or commit persistence. The vacuous in-memory restart-filter test was removed. Docker Kafka compose setups exist in the repository, but no automated restart/offset test harness was found; no broker-backed test was added or run. Existing tenant-policy and startup-validation tests remain.

Validation

On trimmed head 176666c3, merged with origin/main at 65ee74ff:

• CFLAGS=-std=gnu17 cargo test --lib --features kafka connectors::kafka: 63 passed, 0 failed, 1 ignored, 1435 filtered out; exit 0. The parent empty_consumer_tenant_env_is_rejected_in_child_process passed, explicitly setting the guard and empty tenant environment value and invoking the ignored child with a 1 passed assertion.
• Explicit ignored child selection with -- --ignored --exact connectors::kafka::config::tenant_env_tests::tenant_env_child: 1 passed, 0 failed, 0 ignored, 1498 filtered out; exit 0. With no child guard set, this direct run returns before asserting; the parent's passing subprocess run supplies the meaningful environment-value check.
• cargo fmt --check: passed, exit 0.
• CFLAGS=-std=gnu17 cargo clippy --features kafka --all-targets: passed, exit 0; 22 diagnostics, 0 new compared with a fresh build of origin/main at 65ee74ff.

Known limitations / scope relative to main

There is no per-chunk processing timeout. A hung chunk stalls that partition's commits: ordered buffering never yields a later commit past an unfinished earlier chunk. For this stall case, no commit passes unprocessed records. On origin/main, for_each_concurrent(None) lets later chunks finish and commit past a hung earlier chunk.

Once all eight buffer positions are occupied behind a hung chunk, the partition stops draining. Its existing 100,000-record queue can fill, and the single consumer thread awaits a send to that full queue, blocking delivery to other partitions and potentially exceeding max.poll.interval.ms. While that send is blocked, the consumer holds the stream-state lock, so a revoke delivered in that window is skipped (skipping rebalance revoke, as on main when the lock is busy); the stale partition task keeps its in-flight chunks, and if they later finish they can commit after the partition has a new owner. That can move the committed offset backwards (duplicates), not past unprocessed records. For an asynchronously hung chunk, main instead keeps draining into unbounded in-flight work, with potential memory growth. Synchronous blocking or a stalled synchronous commit can impede progress in either implementation.

Processing-failure and revoke policies are unchanged from main: ordinary failed chunks still advance the committed offset without retry, and there is no post-revoke generation fence for commits. The separate scaling stack's layer 08 (stack/08-kafka-commit) supplies a serial run_partition, retries, partition stop, and a commit gate on the contiguous durable frontier; 08b (stack/08b-kafka-durable-ack) extends durable acknowledgments. Bounded timeout/recovery work belongs to that stack's 08/08b scope. The current refs have a bounded revoke-drain timeout, not a per-chunk processing deadline. These layers are separate from #140 and do not yet include it.

Panic handling is unchanged from main. An async panic while polling process, including prepare_stream, unwinds the partition task under either buffered(8) or for_each_concurrent(None); neither converts it into an ordinary processing error. The panicking chunk's own commit callback is not reached. Main may already have committed later chunks before that panic; ordered buffering prevents that while the earlier chunk remains unfinished. In both versions, the consumer retains the closed partition sender, continues fetching, and logs/drops subsequent records in memory without committing them until revoke or restart. A panic inside the spawned Rayon closure follows Rayon's default process-abort policy; no custom panic handler is installed.

Records already committed after tenant rejection are not replayed when configuration is corrected. Recovery requires a manual consumer-group offset reset and may duplicate data already ingested.

Per-topic tenant routing is not supported. Configured tenant IDs are validated syntactically but are not checked for existence or ingestion/workspace suspension; that pre-existing issue is tracked as a follow-up.

Restart coverage is unit-level, not broker-backed: tests exercise next-offset calculation and ordered commits through an injected callback, not broker persistence, restart, or reassignment.

🤖 Generated with Claude Code