Kafka gateway: accept the topic configurations that frameworks send #4508

Follow-up to #3546.

Summary

CreateTopics answers INVALID_CONFIG (40) to any topic configuration, and DescribeConfigs reports only retention.ms. Kafka frameworks create their own topics with configurations, and several of them read the configurations back. Today they stop at startup or hang.

Who hits it, with defaults:

• Kafka Connect in distributed mode creates three internal topics with cleanup.policy=compact and replication factor 3.
• Kafka Streams sets message.timestamp.type on every internal topic and cleanup.policy=compact on changelogs. An app with a state store stops at its first rebalance.
• Schema Registry creates _schemas with compact and tests the policy. Apicurio Registry 3 creates topics with retention.bytes=-1 and reads that key back.
• Debezium creates its schema history topic with retention.ms and retention.bytes.
• MirrorMaker 2 creates its target topics with replication factor 2 and the source configurations. The Connect dead letter queue uses replication factor 3.
• kafka-ui, AKHQ, Terraform and JulieOps send configurations on create. The Strimzi Topic Operator describes a BROKER resource at startup.
• kafka-topics --under-min-isr-partitions throws a NullPointerException without min.insync.replicas.

#4398 made Connect worse. DescribeConfigs reports cleanup.policy=delete. Connect refuses an internal topic with any policy other than compact, even a topic that an operator created by hand. When the entry is missing, Connect logs one INFO line and starts.

Open questions

1. Report cleanup.policy=compact before Iggy compacts? Proposal: yes. Store the policy per topic, and give a compact topic no expiry, so it never loses data. Compaction itself is Kafka gateway: log compaction for compacted topics #4521.
2. Replication factor: accept any value and report the replica count of the Iggy cluster, or accept only values up to that count?
3. Kafka topic configurations that Iggy cannot honor: store and echo them, accept and ignore them as WarpStream does, or refuse them?

Scope

• Stop reporting cleanup.policy=delete until the gateway stores a policy per topic, so Connect starts on topics created by hand
• Accept -1 partitions and replication factor at every CreateTopics version. Kafka's controller accepts them at any version, and KafkaJS createTopics() sends them at v3
• Accept every valid Kafka topic configuration name. Answer 40 only for an unknown name or a bad value
• Map retention.ms to message_expiry, retention.bytes to max_topic_size times the partition count, and segment.bytes to segment_size
• Store cleanup.policy per topic as an Iggy topic option, and report the stored value (open question 1)
• Handle min.insync.replicas, max.message.bytes, message.timestamp.type, unclean.leader.election.enable, compression.type and the compaction keys as open question 3 decides
• Accept a replication factor above 1 as open question 2 decides
• Return the full default set from DescribeConfigs, with source DEFAULT_CONFIG
• Answer a BROKER resource with fixed values, for example auto.create.topics.enable, default.replication.factor, num.partitions, delete.topic.enable and log.retention.ms. Leave out inter.broker.protocol.version until Kafka gateway: serve IncrementalAlterConfigs (key 44) #4509 lands, because kafka-ui picks its alter API from it
• Update gateways/kafka/docs/CONFIGS.md

Acceptance criteria

• A Kafka Connect distributed worker creates its own internal topics through the gateway and starts
• Schema Registry starts with its default topic validation
• A Kafka Streams app with a state store gets past its first rebalance
• Apicurio Registry 3 with KafkaSQL storage starts
• The Strimzi Topic Operator starts without STRIMZI_SKIP_CLUSTER_CONFIG_REVIEW=true

References

• Connect policy test: TopicAdmin.java:527-552. Connect treats 40 as fatal: TopicAdmin.java:443-446
• Connect internal topic defaults: DistributedConfig.java:462-466
• Streams internal topic configurations: InternalTopicConfig.java:38-41, InternalTopicManager.java:563-593
• MirrorMaker 2: MirrorSourceConfig.java:35-37
• Dead letter queue: SinkConnectorConfig.java:65
• -1 at any version: ReplicationControlManager.java:782-792. The Java client refuses to send it below v4: CreateTopicsRequest.java:46-57
• --under-min-isr-partitions: TopicCommand.java:314-316
• Schema Registry KafkaStore.java, Apicurio KafkaAdminUtil.java, Debezium KafkaSchemaHistory.java, Strimzi BatchingTopicController.java
• WarpStream: https://docs.warpstream.com/warpstream/reference/protocol-and-feature-support