Mohith26 opened a new pull request, #18837: URL: https://github.com/apache/iceberg/pull/18837
Every task's control consumer reads every event on the control topic, including the other tasks' `DATA_WRITTEN` replies. Those carry the full data file metadata for a commit and are by far the largest events, yet `Worker` decodes each one only to drop it in `receive`. With many tasks this decode work grows with the square of the task count and runs on the task thread. Eon's ingest-bench write-up of a 192-task run traces commit requests arriving after the commit timeout partly to this ([attempted-engines.md](https://github.com/eon-io/ingest-bench/blob/main/docs/attempted-engines.md#observed-bottlenecks)). This change: - adds two record headers to every control event, `iceberg.connect.group-id` and `iceberg.connect.payload-type` - adds `Channel.handles(PayloadType)`, overridden by `Worker` (`START_COMMIT`) and `Coordinator` (`DATA_WRITTEN`, `DATA_COMPLETE`) to match what their `receive` acts on - skips a record before decoding when its group header names another connector or its type header names a type the channel does not handle. The control-topic offset still advances as before. Records without the headers, or with a type name this version does not know, are decoded exactly as today, so a rolling upgrade with mixed versions keeps working. Older readers ignore the headers. Fetch volume is unchanged; this only removes the decode. **Test plan** - headers are written on sent events (`TestChannel`) - events for another connector, and event types a channel does not handle, are skipped without decoding: their values are deliberately not valid Avro, so the tests fail if they are decoded, and offsets still advance (`TestChannel`, `TestWorker`) - events without headers and events with an unknown type name are still decoded (`TestChannel`) - a record produced by `send` round-trips through a reader that skips its type (`TestChannel`) The new skip tests fail with the skip disabled. `./gradlew :iceberg-kafka-connect:iceberg-kafka-connect:check` passes (173 tests). **Benchmark** (JDK 17, Apple Silicon arm64; code: https://gist.github.com/Mohith26/c7ec146b57f6b0f5a58729a51b55975a). A `DATA_WRITTEN` reply with 512 data files and metrics for 100 columns (the default inferred-metrics limit) encodes to 2.0 MB, close to the ~1.5 MB replies in the run above. | | median | |---|---:| | decode one reply | 46.9 ms | | worker consumes 48 replies + 1 `START_COMMIT`, no headers (decode all) | 2078.6 ms | | same, with headers (skip) | 0.4 ms | 7 runs after 2 warm-ups, in-process `MockConsumer`, so this measures decode cost only, not fetch. This change was written with AI assistance; the tests and benchmark above were run locally before opening. The part worth a closer look is keeping `handles` in sync with `receive` in the two subclasses. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
