Deal with topic recreation for producer and consumer (using metadata)

[milindl]

There is a bunch of things we need to do for producers when topics are recreated (topic id changes). Unfortunately we cannot do it all because the Produce RPC doesn't yet use topic ids. We can make a best-effort from metadata requests.

Similar to produce, fetch also has to be changed. We make these changes from metadata, luckily fetch has UNKNOWN_TOPIC_ID errors, so that helps us be more deterministic rather than best-effort.

Here's a summary of changes:

1. reset leader epoch on topic recreate
2. producer: drain and bump epoch for idempotent/tx producer
3. consumer: reset offset for owned partitions of recreated topic

This adds mock broker changes, mock tests, and real broker tests (only for consumer, it's not yet possible to have real-broker tests for producer as it's not using topic ids on the broker)

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[Copilot]

Copilot reviewed 6 out of 8 changed files in this pull request and generated 1 comment.

Files not reviewed (2)

• tests/CMakeLists.txt: Language not supported
• win32/tests/tests.vcxproj: Language not supported

Comments suppressed due to low confidence (1)

src/rdkafka_topic.c:1010

• The variable 'is_idempodent' is misspelled; please rename it to 'is_idempotent' for clarity.

if (is_idempodent && rd_kafka_pid_valid(rktp->rktp_eos.pid) && !topic_id_change)

[Copilot]

Copilot reviewed 7 out of 9 changed files in this pull request and generated no comments.

Files not reviewed (2)

• tests/CMakeLists.txt: Language not supported
• win32/tests/tests.vcxproj: Language not supported

Comments suppressed due to low confidence (1)

src/rdkafka_topic.c:1044

• The variable 'is_idempodent' appears to be misspelled. It should likely be renamed to 'is_idempotent' for consistency.

if (is_idempodent && rd_kafka_pid_valid(rktp->rktp_eos.pid) && !topic_id_change)

[Copilot]

Copilot reviewed 8 out of 10 changed files in this pull request and generated no comments.

Files not reviewed (2)

• tests/CMakeLists.txt: Language not supported
• win32/tests/tests.vcxproj: Language not supported

Comments suppressed due to low confidence (2)

src/rdkafka_topic.c:1044

• Possible typo in variable name 'is_idempodent'; consider renaming it to 'is_idempotent' for clarity.

if (is_idempodent && rd_kafka_pid_valid(rktp->rktp_eos.pid) && !topic_id_change)

tests/0107-topic_recreate.c:325

• The producer handle 'rk_producer' is NULL at this point; ensure a valid handle is provided when calling test_delete_topic.

test_delete_topic(rk_producer, topic);