This is an automated email from the ASF dual-hosted git repository. krishvishal pushed a commit to branch kafka-configs-regression in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 82c050dad63694bc6df6cac2f4adf176f3eefce3 Author: Krishna Vishal <[email protected]> AuthorDate: Tue Oct 13 12:10:27 2026 +0530 fix(gateways): stop reporting cleanup.policy and accept -1 defaults Connect refuses its topics when cleanup.policy is delete, and KafkaJS createTopics() sends -1 at v3. --- gateways/kafka/docs/CONFIGS.md | 31 +++-- gateways/kafka/docs/MANUAL_TESTING.md | 4 +- gateways/kafka/docs/SCOPE.md | 4 +- .../kafka/src/protocol/handlers/create_topics.rs | 125 +++++++-------------- .../src/protocol/handlers/describe_configs.rs | 42 +++---- .../kafka/src/protocol/handlers/topic_config.rs | 55 +++++---- gateways/kafka/tests/configs_java_client_tests.rs | 5 +- gateways/kafka/tests/configs_real_bridge_tests.rs | 18 +-- .../kafka/tests/create_topics_real_bridge_tests.rs | 45 ++++++++ gateways/kafka/tests/java/ConfigsProbe.java | 17 +-- gateways/kafka/tests/response_negative_tests.rs | 11 +- 11 files changed, 175 insertions(+), 182 deletions(-) diff --git a/gateways/kafka/docs/CONFIGS.md b/gateways/kafka/docs/CONFIGS.md index 19fa2e570..8dd1a3f4d 100644 --- a/gateways/kafka/docs/CONFIGS.md +++ b/gateways/kafka/docs/CONFIGS.md @@ -23,20 +23,26 @@ authorization check of its own. ## What a describe returns -A null or empty `configuration_keys` list returns both keys below. A non-empty +A null or empty `configuration_keys` list returns `retention.ms`. A non-empty list returns only the requested names this gateway knows, in request order, with a name repeated in the list collapsed to one entry. A name it does not -know, including `retention.minutes` and `retention.hours`, is omitted. The -resource's error code stays `NONE` (0), and the known keys in that request are -still returned. +know, including `cleanup.policy`, `retention.minutes` and `retention.hours`, is +omitted. The resource's error code stays `NONE` (0), and the known keys in that +request are still returned. | Key | Value | Read only | Source | | --- | ----- | --------- | ------ | | `retention.ms` | Sealed-segment lifetime from Iggy `message_expiry`, or `-1` when those segments never expire | no | `DEFAULT_CONFIG` (5) for the unset never-expire default; `DYNAMIC_TOPIC_CONFIG` (1) when a client set it or a stored duration cannot be shown as milliseconds | -| `cleanup.policy` | `delete` | yes | `DEFAULT_CONFIG` (5) | -`is_sensitive` is false for both. From v3, `include_documentation` is honored: -true attaches one sentence per known key, false sends no documentation. +The gateway does not report `cleanup.policy` until it stores a policy per +topic. Iggy deletes expired messages and does not compact, so the true value is +`delete`. Kafka Connect refuses an internal topic that reports any policy other +than `compact`. When the entry is missing, Connect starts +(`TopicAdmin.verifyTopicCleanupPolicyOnlyCompact`). A reported `delete` +therefore stops Connect, even on topics that an operator created for it. + +`is_sensitive` is false. From v3, `include_documentation` is honored: true +attaches one sentence per known key, false sends no documentation. `include_synonyms` has no effect: this gateway models no config-inheritance chain beyond `retention.ms` itself, so the synonym list on every entry is always empty. @@ -86,11 +92,14 @@ call `update_topic`. ## Limitations -- `IncrementalAlterConfigs` (API 44) is not implemented here. It is planned - after this work merges. It is not advertised, so a client falls back to - `AlterConfigs` (33); see "Authorization" above for what an unadvertised key - does to the client. +- `IncrementalAlterConfigs` (API 44) is not implemented and is not + advertised. No client falls back to `AlterConfigs` (33) on its own. The Java + client and librdkafka fail locally, as the top of this page says. Kafka 4.0 + removed `Admin.alterConfigs`, and `kafka-configs --alter` sends only API 44. + A Java 4.x client or tool therefore cannot change `retention.ms` through this + gateway. - `retention.ms` remains the only value Iggy stores. +- `cleanup.policy` is not reported. See "What a describe returns". ## Other resource errors diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md index 36a6602cc..21c88dd13 100644 --- a/gateways/kafka/docs/MANUAL_TESTING.md +++ b/gateways/kafka/docs/MANUAL_TESTING.md @@ -387,8 +387,8 @@ criteria). | 21 | INVALID_REQUIRED_ACKS | Produce with a bridge: `acks` is not 0, 1 or -1 | | 29 | TOPIC_AUTHORIZATION_FAILED | Produce with a bridge: Iggy user lacks permission | | 35 | UNSUPPORTED_VERSION | Out of range: **ApiVersions only** (KIP-511 exception); every other API key's out-of-range version closes the connection instead - see Category B. In range: InitProducerId with a `transactional_id`; Produce with a `transactional_id`, on every partition with or without a bridge (an `acks=0` one closes instead); and, with a bridge, a transactional or control batch - transactions are not supported, see `SCOPE.md` | -| 37 | INVALID_PARTITIONS | CreateTopics: partition count `0` or `< -1` (or any non-positive on v2–v3) | -| 38 | INVALID_REPLICATION_FACTOR | CreateTopics: replication factor `0` or `< -1` (or any non-positive on v2–v3) | +| 37 | INVALID_PARTITIONS | CreateTopics: partition count `0` or `< -1` | +| 38 | INVALID_REPLICATION_FACTOR | CreateTopics: replication factor `0` or `< -1` | | 41 | NOT_CONTROLLER | CreateTopics stub (topic not created) | | 42 | INVALID_REQUEST | Fetch/ListOffsets/CreateTopics/ApiVersions decode failure. Not Produce: it closes the connection (H1) | | 76 | UNSUPPORTED_COMPRESSION_TYPE | Produce with a bridge: zstd before v7 | diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md index 91d2e7e36..ea1eec3d6 100644 --- a/gateways/kafka/docs/SCOPE.md +++ b/gateways/kafka/docs/SCOPE.md @@ -54,7 +54,7 @@ it knows the server supports flexible encoding. | 0 | Produce | 3 | 9 | 3, 4, 5, 6, 7, 8, 9 | With a bridge: one `send_messages` per partition. Without one: stub returns `NOT_LEADER_OR_FOLLOWER` (6) | | 1 | Fetch | 4 | 12 | 4, 5, 6, 7, 8, 9, 10, 11, 12 | With a bridge: shared topic probes, paged `poll_messages` per partition with records, wait up to `max_wait_ms` (max 18 s). Without one: stub returns `NOT_LEADER_OR_FOLLOWER` (6) | | 2 | ListOffsets | 1 | 6 | 1, 2, 3, 4, 5, 6 | Decode request; stub response | -| 19 | CreateTopics | 2 | 5 | 2, 3, 4, 5 | Decode request; stub returns `NOT_CONTROLLER` (41); `-1` partitions/RF = broker default on v4+ | +| 19 | CreateTopics | 2 | 5 | 2, 3, 4, 5 | Decode request; stub returns `NOT_CONTROLLER` (41); `-1` partitions/RF = broker default at every version | | 20 | DeleteTopics | 1 | 5 | 1, 2, 3, 4, 5 | With a bridge: deletes the Iggy topic, never the backing stream (see `BRIDGE_MAPPING.md`). Without one: stub returns `NOT_CONTROLLER` (41). v6 (topic-id-based) not advertised - this bridge has no topic-id concept | | 8 | OffsetCommit | 2 | 9 | 2 … 9 | With a bridge: stores each offset as an Iggy external group offset, and a negative offset deletes it ([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)). Without one: `COORDINATOR_LOAD_IN_PROGRESS` (14). Flexible encoding at v8+ | | 9 | OffsetFetch | 1 | 9 | 1 … 9 | With a bridge: reads each offset, -1 when none is stored, and a null topic list reads every topic that holds the group. Without one: 14. Flexible encoding at v6+ | @@ -66,7 +66,7 @@ it knows the server supports flexible encoding. | 15 | DescribeGroups | 0 | 6 | 0, 1, 2, 3, 4, 5, 6 | Members, assignment and state of groups on this coordinator; a missing group is `Dead`, and v6 also returns `GROUP_ID_NOT_FOUND` (69); flexible encoding at v5+ | | 16 | ListGroups | 0 | 5 | 0, 1, 2, 3, 4, 5 | Groups on this coordinator; v4+ state filter and v5 type filter; flexible encoding at v3+ | | 22 | InitProducerId | 0 | 5 | 0, 1, 2, 3, 4, 5 | Allocate a producer id (epoch 0); a `transactional_id` gets `UNSUPPORTED_VERSION` (35); flexible encoding at v2+ | -| 32 | DescribeConfigs | 1 | 4 | 1, 2, 3, 4 | Topic resources only. Returns `retention.ms` and `cleanup.policy`. Flexible encoding at v4. See [`CONFIGS.md`](CONFIGS.md) | +| 32 | DescribeConfigs | 1 | 4 | 1, 2, 3, 4 | Topic resources only. Returns `retention.ms` and leaves out `cleanup.policy`. Flexible encoding at v4. See [`CONFIGS.md`](CONFIGS.md) | | 33 | AlterConfigs | 0 | 2 | 0, 1, 2 | Persists `retention.ms` only. Flexible encoding at v2. See [`CONFIGS.md`](CONFIGS.md) | A request is accepted when `min_version ≤ api_version ≤ max_version` for that API key. Any other version for a listed key closes the connection (ApiVersions excepted - see Governance model above). diff --git a/gateways/kafka/src/protocol/handlers/create_topics.rs b/gateways/kafka/src/protocol/handlers/create_topics.rs index d43c09d10..83ce99395 100644 --- a/gateways/kafka/src/protocol/handlers/create_topics.rs +++ b/gateways/kafka/src/protocol/handlers/create_topics.rs @@ -136,7 +136,6 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand let deadline = Instant::now() + clamp_request_timeout(req.timeout_ms); let results = create_all_topics( bridge, - api_version, &req.topics, &duplicate_names, req.validate_only, @@ -166,7 +165,6 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand /// call itself, the same pattern `list_offsets.rs` uses, not rely on `timeout_at` to skip it. async fn create_all_topics( bridge: &IggyBridge, - api_version: i16, topics: &[CreatableTopic], duplicate_names: &HashSet<&str>, validate_only: bool, @@ -201,12 +199,8 @@ async fn create_all_topics( continue; } - let result = match tokio::time::timeout_at( - deadline, - create_one_topic(bridge, api_version, topic, validate_only), - ) - .await - { + let create = create_one_topic(bridge, topic, validate_only); + let result = match tokio::time::timeout_at(deadline, create).await { Ok(result) => result, Err(_elapsed) => { deadline_exceeded = true; @@ -247,7 +241,6 @@ async fn create_all_topics( /// one caller does. async fn create_one_topic( bridge: &IggyBridge, - version: i16, topic: &CreatableTopic, validate_only: bool, ) -> CreatableTopicResult { @@ -256,7 +249,7 @@ async fn create_one_topic( .with_error_message(None); let kafka_topic = topic.name.as_str(); - let partition_count = match local_shape_error(version, topic) { + let partition_count = match local_shape_error(topic) { Ok(count) => count, Err((error_code, error_message)) => { return match bridge.get_kafka_topic(kafka_topic).await { @@ -308,10 +301,7 @@ async fn create_one_topic( /// once, only on the rejection path. Returns `(error_code, error_message)` rather than a whole /// [`CreatableTopicResult`] to keep this `Result`'s `Err` arm small (`clippy::result_large_err`); /// the caller already holds the shared `name`/base fields to rebuild the full result from. -fn local_shape_error( - version: i16, - topic: &CreatableTopic, -) -> core::result::Result<u32, (i16, Option<StrBytes>)> { +fn local_shape_error(topic: &CreatableTopic) -> core::result::Result<u32, (i16, Option<StrBytes>)> { if !topic.configs.is_empty() { return Err(( ERROR_INVALID_CONFIG, @@ -321,7 +311,7 @@ fn local_shape_error( )), )); } - validate_create_topic_shape(version, topic).map_err(|code| (code, None)) + validate_create_topic_shape(topic).map_err(|code| (code, None)) } /// Maps a bridge failure to a Kafka result, logging the real cause server-side. @@ -405,12 +395,10 @@ fn bridge_error_result( /// binding this way, not just the index. /// /// With no assignments, `num_partitions = -1` / `replication_factor = -1` mean "use the broker -/// default" from v4+ (pre-v4 requires an explicit positive value for both, since v2/v3 have no -/// broker-default sentinel absent a manual assignment). -fn validate_create_topic_shape( - version: i16, - topic: &CreatableTopic, -) -> core::result::Result<u32, i16> { +/// default" at every version. The Java client refuses to send them below v4, but Kafka's +/// controller never checks the version (`ReplicationControlManager.createTopic`), and +/// `KafkaJS` `createTopics()` sends both at v3. +fn validate_create_topic_shape(topic: &CreatableTopic) -> core::result::Result<u32, i16> { if !topic.assignments.is_empty() { if topic.num_partitions != -1 || topic.replication_factor != -1 { return Err(ERROR_INVALID_REQUEST); @@ -423,9 +411,7 @@ fn validate_create_topic_shape( return Ok(u32::try_from(topic.assignments.len()).unwrap_or(DEFAULT_PARTITION_COUNT)); } - let broker_default_ok = version >= 4; - - // Checked ahead of (and separately from) `partitions_ok` below: that check's own + // Checked ahead of (and separately from) the partition-count check below: that check's own // `INVALID_PARTITIONS` (37) means "count is below 1" on the wire (`bridge/error.rs`'s own // comment on `IggyError::TooManyPartitions`), the opposite condition from "too many" - reusing // it here would send a client-visible message that contradicts the request it just sent. This @@ -436,12 +422,7 @@ fn validate_create_topic_shape( return Err(ERROR_INVALID_REQUEST); } - let partitions_ok = if broker_default_ok { - topic.num_partitions == -1 || topic.num_partitions > 0 - } else { - topic.num_partitions > 0 - }; - if !partitions_ok { + if topic.num_partitions != -1 && topic.num_partitions <= 0 { return Err(ERROR_INVALID_PARTITIONS); } @@ -451,12 +432,7 @@ fn validate_create_topic_shape( // controller rejects `replication_factor > 1` the same way (`ReplicationControlManager`) - // accepting it here and silently reporting back `1` (`success()`, below) would tell the // client its request succeeded as sent when it didn't. - let replication_ok = if broker_default_ok { - topic.replication_factor == -1 || topic.replication_factor == 1 - } else { - topic.replication_factor == 1 - }; - if !replication_ok { + if topic.replication_factor != -1 && topic.replication_factor != 1 { return Err(ERROR_INVALID_REPLICATION_FACTOR); } @@ -523,36 +499,20 @@ pub fn encode_response(version: i16, req: &CreateTopicsRequest) -> Result<Bytes> /// Resolve per-topic `CreateTopics` error for the stub (no-bridge) path. /// -/// KIP-464: `num_partitions = -1` / `replication_factor = -1` mean broker default when either -/// (a) the version is v4+, or (b) the topic carries a manual partition assignment (valid on -/// v2/v3 as well). Otherwise non-positive values are [`ERROR_INVALID_PARTITIONS`] / -/// [`ERROR_INVALID_REPLICATION_FACTOR`]. When validation passes, the stub returns -/// [`ERROR_NOT_CONTROLLER`] so clients do not believe the topic was created. -const fn topic_error(version: i16, topic: &CreatableTopic, forced_error: i16) -> i16 { +/// KIP-464: `num_partitions = -1` / `replication_factor = -1` mean broker default at every +/// version, as in [`validate_create_topic_shape`]. Other non-positive values are +/// [`ERROR_INVALID_PARTITIONS`] / [`ERROR_INVALID_REPLICATION_FACTOR`]. When validation passes, +/// the stub returns [`ERROR_NOT_CONTROLLER`] so clients do not believe the topic was created. +const fn topic_error(topic: &CreatableTopic, forced_error: i16) -> i16 { if forced_error != ERROR_NONE { return forced_error; } - - let broker_default_ok = version >= 4 || !topic.assignments.is_empty(); - - let partitions_ok = if broker_default_ok { - topic.num_partitions == -1 || topic.num_partitions > 0 - } else { - topic.num_partitions > 0 - }; - if !partitions_ok { + if topic.num_partitions != -1 && topic.num_partitions <= 0 { return ERROR_INVALID_PARTITIONS; } - - let replication_ok = if broker_default_ok { - topic.replication_factor == -1 || topic.replication_factor > 0 - } else { - topic.replication_factor > 0 - }; - if !replication_ok { + if topic.replication_factor != -1 && topic.replication_factor <= 0 { return ERROR_INVALID_REPLICATION_FACTOR; } - ERROR_NOT_CONTROLLER } @@ -562,7 +522,7 @@ fn encode_inner(version: i16, topics: &[CreatableTopic], forced_error: i16) -> R .map(|topic| { CreatableTopicResult::default() .with_name(topic.name.clone()) - .with_error_code(topic_error(version, topic, forced_error)) + .with_error_code(topic_error(topic, forced_error)) .with_error_message(None) .with_num_partitions(topic.num_partitions) .with_replication_factor(topic.replication_factor) @@ -607,31 +567,26 @@ mod tests { #[test] fn explicit_partitions_and_replication_resolve_as_given() { let topic = creatable_topic(3, 1); - assert_eq!(validate_create_topic_shape(2, &topic), Ok(3)); + assert_eq!(validate_create_topic_shape(&topic), Ok(3)); } #[test] - fn v4_plus_accepts_broker_default_sentinel_with_no_assignments() { - let topic = creatable_topic(-1, -1); + fn broker_default_sentinels_with_no_assignments_resolve_to_the_default_count() { assert_eq!( - validate_create_topic_shape(4, &topic), + validate_create_topic_shape(&creatable_topic(-1, -1)), Ok(DEFAULT_PARTITION_COUNT) ); - } - - #[test] - fn pre_v4_rejects_broker_default_sentinel_with_no_assignments() { - let topic = creatable_topic(-1, -1); assert_eq!( - validate_create_topic_shape(2, &topic), - Err(ERROR_INVALID_PARTITIONS) + validate_create_topic_shape(&creatable_topic(-1, 1)), + Ok(DEFAULT_PARTITION_COUNT) ); + assert_eq!(validate_create_topic_shape(&creatable_topic(4, -1)), Ok(4)); } #[test] - fn pre_v4_accepts_broker_default_sentinel_with_a_manual_assignment() { + fn broker_default_sentinels_with_a_manual_assignment_take_its_length() { let topic = creatable_topic(-1, -1).with_assignments(vec![assignment(0), assignment(1)]); - assert_eq!(validate_create_topic_shape(2, &topic), Ok(2)); + assert_eq!(validate_create_topic_shape(&topic), Ok(2)); } #[test] @@ -639,7 +594,7 @@ mod tests { let topic = creatable_topic(-1, -1).with_assignments(vec![assignment_with_replicas(0, vec![])]); assert_eq!( - validate_create_topic_shape(2, &topic), + validate_create_topic_shape(&topic), Err(ERROR_INVALID_REPLICA_ASSIGNMENT) ); } @@ -649,7 +604,7 @@ mod tests { let topic = creatable_topic(-1, -1).with_assignments(vec![assignment_with_replicas(0, vec![1, 1])]); assert_eq!( - validate_create_topic_shape(2, &topic), + validate_create_topic_shape(&topic), Err(ERROR_INVALID_REPLICA_ASSIGNMENT) ); } @@ -661,7 +616,7 @@ mod tests { let topic = creatable_topic(-1, -1).with_assignments(vec![assignment_with_replicas(0, vec![7])]); assert_eq!( - validate_create_topic_shape(2, &topic), + validate_create_topic_shape(&topic), Err(ERROR_INVALID_REPLICA_ASSIGNMENT) ); } @@ -670,7 +625,7 @@ mod tests { fn zero_partitions_is_rejected_regardless_of_version() { let topic = creatable_topic(0, 1); assert_eq!( - validate_create_topic_shape(5, &topic), + validate_create_topic_shape(&topic), Err(ERROR_INVALID_PARTITIONS) ); } @@ -679,7 +634,7 @@ mod tests { fn zero_replication_factor_is_rejected_regardless_of_version() { let topic = creatable_topic(1, 0); assert_eq!( - validate_create_topic_shape(5, &topic), + validate_create_topic_shape(&topic), Err(ERROR_INVALID_REPLICATION_FACTOR) ); } @@ -692,7 +647,7 @@ mod tests { // accepted-and-silently-downgraded. let topic = creatable_topic(1, 3); assert_eq!( - validate_create_topic_shape(5, &topic), + validate_create_topic_shape(&topic), Err(ERROR_INVALID_REPLICATION_FACTOR) ); } @@ -700,14 +655,14 @@ mod tests { #[test] fn replication_factor_of_exactly_one_is_accepted() { let topic = creatable_topic(1, 1); - assert_eq!(validate_create_topic_shape(5, &topic), Ok(1)); + assert_eq!(validate_create_topic_shape(&topic), Ok(1)); } #[test] fn partition_count_at_the_cap_is_accepted() { let topic = creatable_topic(i32::try_from(MAX_PARTITIONS_COUNT).unwrap(), 1); assert_eq!( - validate_create_topic_shape(5, &topic), + validate_create_topic_shape(&topic), Ok(MAX_PARTITIONS_COUNT) ); } @@ -720,7 +675,7 @@ mod tests { // (`bridge/error.rs::too_many_partitions_maps_to_invalid_request_not_invalid_partitions`). let topic = creatable_topic(i32::try_from(MAX_PARTITIONS_COUNT).unwrap() + 1, 1); assert_eq!( - validate_create_topic_shape(5, &topic), + validate_create_topic_shape(&topic), Err(ERROR_INVALID_REQUEST) ); } @@ -734,7 +689,7 @@ mod tests { // which one would win if it did. let topic = creatable_topic(2, 1).with_assignments(vec![assignment(0), assignment(1)]); assert_eq!( - validate_create_topic_shape(5, &topic), + validate_create_topic_shape(&topic), Err(ERROR_INVALID_REQUEST) ); } @@ -743,7 +698,7 @@ mod tests { fn explicit_num_partitions_disagreeing_with_assignments_length_is_also_rejected() { let topic = creatable_topic(3, 1).with_assignments(vec![assignment(0), assignment(1)]); assert_eq!( - validate_create_topic_shape(5, &topic), + validate_create_topic_shape(&topic), Err(ERROR_INVALID_REQUEST) ); } @@ -753,7 +708,7 @@ mod tests { { let topic = creatable_topic(-1, 1).with_assignments(vec![assignment(0), assignment(1)]); assert_eq!( - validate_create_topic_shape(5, &topic), + validate_create_topic_shape(&topic), Err(ERROR_INVALID_REQUEST) ); } diff --git a/gateways/kafka/src/protocol/handlers/describe_configs.rs b/gateways/kafka/src/protocol/handlers/describe_configs.rs index 54017c66e..90ac2dc92 100644 --- a/gateways/kafka/src/protocol/handlers/describe_configs.rs +++ b/gateways/kafka/src/protocol/handlers/describe_configs.rs @@ -39,8 +39,7 @@ use crate::protocol::api::{ }; use crate::protocol::bounds_guard::validate_describe_configs_shape; use crate::protocol::handlers::topic_config::{ - CLEANUP_DOC, CLEANUP_POLICY, CLEANUP_POLICY_VALUE, CONFIG_SOURCE_DEFAULT, CONFIG_SOURCE_TOPIC, - CONFIG_TYPE_LONG, CONFIG_TYPE_STRING, ListedKey, ONLY_TOPIC_RESOURCES, RESOURCE_TYPE_TOPIC, + CONFIG_SOURCE_TOPIC, CONFIG_TYPE_LONG, ListedKey, ONLY_TOPIC_RESOURCES, RESOURCE_TYPE_TOPIC, RETENTION_DOC, RETENTION_MS, bridge_failure, listed_keys, name_reason, retention_ms, topic_cap_and_duplicates, topic_cap_message, }; @@ -289,14 +288,6 @@ fn described_entries( include_documentation.then_some(RETENTION_DOC), )); } - ListedKey::Cleanup => configs.push(config_entry( - CLEANUP_POLICY, - Some(CLEANUP_POLICY_VALUE), - true, - CONFIG_SOURCE_DEFAULT, - CONFIG_TYPE_STRING, - include_documentation.then_some(CLEANUP_DOC), - )), // A resource error makes clients drop every returned config, including // retention.ms. An unmodeled name is omitted and the resource stays successful. ListedKey::Unknown => {} @@ -346,9 +337,7 @@ mod tests { CompressionAlgorithm, HeaderKey, HeaderValue, IggyByteSize, IggyExpiry, IggyTimestamp, MaxTopicSize, OptionValue, ResourceOptions, TopicDetails, topic_option_keys, }; - use kafka_protocol::messages::describe_configs_response::{ - DescribeConfigsResourceResult, DescribeConfigsSynonym, - }; + use kafka_protocol::messages::describe_configs_response::DescribeConfigsResourceResult; use kafka_protocol::protocol::StrBytes; use super::described_entries; @@ -401,7 +390,7 @@ mod tests { } #[test] - fn default_describe_returns_retention_and_cleanup() { + fn default_describe_returns_retention_only() { let expiry = parse_retention_ms("120000").expect("2 minutes in ms"); let details = topic_details(expiry, true); let (configs, bad) = entries(&details, None, false); @@ -411,14 +400,27 @@ mod tests { .iter() .map(|config| config.name.as_str()) .collect::<Vec<_>>(), - vec![RETENTION_MS, CLEANUP_POLICY] + vec![RETENTION_MS] ); assert_eq!(value_of(&configs, RETENTION_MS), Some("120000")); - let cleanup = configs - .iter() - .find(|config| config.name.as_str() == CLEANUP_POLICY) - .expect("cleanup"); - assert_eq!(cleanup.synonyms, [] as [DescribeConfigsSynonym; 0]); + } + + #[test] + fn cleanup_policy_is_omitted_even_when_asked_for() { + let details = topic_details(IggyExpiry::NeverExpire, false); + let only_cleanup = [StrBytes::from_static_str(CLEANUP_POLICY)]; + let (configs, bad) = entries(&details, Some(&only_cleanup), true); + assert!(!bad); + assert_eq!(configs, [] as [DescribeConfigsResourceResult; 0]); + + let both = [ + StrBytes::from_static_str(CLEANUP_POLICY), + StrBytes::from_static_str(RETENTION_MS), + ]; + let (configs, bad) = entries(&details, Some(&both), true); + assert!(!bad); + assert_eq!(configs.len(), 1); + assert_eq!(configs[0].name.as_str(), RETENTION_MS); } #[test] diff --git a/gateways/kafka/src/protocol/handlers/topic_config.rs b/gateways/kafka/src/protocol/handlers/topic_config.rs index 9da428bc5..1c1afc243 100644 --- a/gateways/kafka/src/protocol/handlers/topic_config.rs +++ b/gateways/kafka/src/protocol/handlers/topic_config.rs @@ -44,20 +44,15 @@ pub const CONFIG_SOURCE_TOPIC: i8 = 1; /// Kafka `ConfigSource.DEFAULT_CONFIG`: the value is the gateway default. pub const CONFIG_SOURCE_DEFAULT: i8 = 5; -/// Kafka `ConfigType.STRING`. -pub const CONFIG_TYPE_STRING: i8 = 2; - /// Kafka `ConfigType.LONG`. pub const CONFIG_TYPE_LONG: i8 = 5; pub const RETENTION_MS: &str = "retention.ms"; pub const CLEANUP_POLICY: &str = "cleanup.policy"; -pub const CLEANUP_POLICY_VALUE: &str = "delete"; pub const ONLY_TOPIC_RESOURCES: &str = "only topic resources are supported"; pub const RETENTION_DOC: &str = "How long a sealed segment is kept, in milliseconds. The active segment does not expire. -1 means sealed segments never expire."; -pub const CLEANUP_DOC: &str = "Iggy deletes expired messages and does not compact a topic."; /// Distinct topic names one `CreateTopics`, `DescribeConfigs`, or `AlterConfigs` request /// may address through the bridge. @@ -73,11 +68,10 @@ pub const MAX_CONFIG_TOPICS: usize = 100; #[derive(Debug, Clone, PartialEq, Eq)] pub enum ListedKey { Retention, - Cleanup, - /// An unmodeled name, including `retention.minutes` and `retention.hours`. The name - /// itself is never read back out: `DescribeConfigs` omits every `Unknown` entry from - /// its response, so carrying the name here would only pay a `String` allocation for a - /// value no caller inspects. + /// An unmodeled name, including `cleanup.policy`, `retention.minutes` and + /// `retention.hours`. The name itself is never read back out: `DescribeConfigs` omits + /// every `Unknown` entry from its response, so carrying the name here would only pay a + /// `String` allocation for a value no caller inspects. Unknown, } @@ -319,21 +313,26 @@ pub fn bridge_failure(error: &BridgeError, handler: &str, action: &str) -> (i16, /// Keys to return for one describe resource. /// -/// `None` and an empty list both mean `retention.ms` and `cleanup.policy`. A name repeated -/// in `configuration_keys` is collapsed to one entry, in its first position. Any other name, -/// including `retention.minutes` and `retention.hours`, is [`ListedKey::Unknown`]. -/// `DescribeConfigs` omits those names from the response and does not fail the resource. +/// `None` and an empty list both mean `retention.ms`. A name repeated in +/// `configuration_keys` is collapsed to one entry, in its first position. Any other name, +/// including `cleanup.policy`, `retention.minutes` and `retention.hours`, is +/// [`ListedKey::Unknown`]. `DescribeConfigs` omits those names from the response and does not +/// fail the resource. +/// +/// `cleanup.policy` stays unknown until the gateway stores a policy per topic. Kafka Connect +/// refuses an internal topic that reports any policy other than `compact`, but it starts when +/// the entry is missing (`TopicAdmin.verifyTopicCleanupPolicyOnlyCompact`). A fixed `delete` +/// therefore stops Connect even on topics an operator created for it. #[must_use] pub fn listed_keys(configuration_keys: Option<&[StrBytes]>) -> Vec<ListedKey> { let Some(keys) = configuration_keys.filter(|keys| !keys.is_empty()) else { - return vec![ListedKey::Retention, ListedKey::Cleanup]; + return vec![ListedKey::Retention]; }; let mut seen = HashSet::new(); keys.iter() .filter(|key| seen.insert(key.as_str())) .map(|key| match key.as_str() { RETENTION_MS => ListedKey::Retention, - CLEANUP_POLICY => ListedKey::Cleanup, _ => ListedKey::Unknown, }) .collect() @@ -421,25 +420,25 @@ mod tests { } #[test] - fn listed_keys_defaults_to_both_and_keeps_an_unknown_name() { - assert_eq!( - listed_keys(None), - vec![ListedKey::Retention, ListedKey::Cleanup] - ); - assert_eq!( - listed_keys(Some(&[])), - vec![ListedKey::Retention, ListedKey::Cleanup] - ); + fn listed_keys_defaults_to_retention_and_keeps_an_unknown_name() { + assert_eq!(listed_keys(None), vec![ListedKey::Retention]); + assert_eq!(listed_keys(Some(&[])), vec![ListedKey::Retention]); let asked = [ - StrBytes::from_static_str(CLEANUP_POLICY), + StrBytes::from_static_str(RETENTION_MS), StrBytes::from_static_str("no.such"), ]; assert_eq!( listed_keys(Some(&asked)), - vec![ListedKey::Cleanup, ListedKey::Unknown] + vec![ListedKey::Retention, ListedKey::Unknown] ); } + #[test] + fn cleanup_policy_is_an_unknown_name_even_when_asked_for() { + let asked = [StrBytes::from_static_str(CLEANUP_POLICY)]; + assert_eq!(listed_keys(Some(&asked)), vec![ListedKey::Unknown]); + } + #[test] fn listed_keys_collapses_a_name_repeated_in_the_request() { let asked = [ @@ -449,7 +448,7 @@ mod tests { ]; assert_eq!( listed_keys(Some(&asked)), - vec![ListedKey::Retention, ListedKey::Cleanup] + vec![ListedKey::Retention, ListedKey::Unknown] ); } diff --git a/gateways/kafka/tests/configs_java_client_tests.rs b/gateways/kafka/tests/configs_java_client_tests.rs index 76ce7bd34..5bd3161a8 100644 --- a/gateways/kafka/tests/configs_java_client_tests.rs +++ b/gateways/kafka/tests/configs_java_client_tests.rs @@ -294,10 +294,7 @@ async fn kafka_39_admin_client_describes_and_alters_topic_configs() { text.contains("retention.ms=-1 source=DEFAULT_CONFIG"), "{text}" ); - assert!( - text.contains("cleanup.policy=delete read_only=true source=DEFAULT_CONFIG"), - "{text}" - ); + assert!(text.contains("cleanup.policy=absent"), "{text}"); assert!( text.contains("unknown_key_rejected=InvalidConfigurationException"), "{text}" diff --git a/gateways/kafka/tests/configs_real_bridge_tests.rs b/gateways/kafka/tests/configs_real_bridge_tests.rs index 7fa62837f..8dbf37a75 100644 --- a/gateways/kafka/tests/configs_real_bridge_tests.rs +++ b/gateways/kafka/tests/configs_real_bridge_tests.rs @@ -152,7 +152,7 @@ fn entry<'a>( #[tokio::test] #[serial] -async fn describe_configs_returns_retention_and_cleanup_and_alter_persists_retention() { +async fn describe_configs_returns_retention_without_cleanup_and_alter_persists_retention() { let data_dir = tempfile::tempdir().expect("tempdir"); let server = TestServer::spawn(data_dir.path()).await; let state = connected_state(&server).await; @@ -168,7 +168,7 @@ async fn describe_configs_returns_retention_and_cleanup_and_alter_persists_reten .await; let result = &described.results[0]; assert_eq!(result.error_code, ERROR_NONE); - assert_eq!(result.configs.len(), 2); + assert_eq!(result.configs.len(), 1); let retention = entry(&result.configs, "retention.ms"); assert_eq!(retention.value.as_ref().map(StrBytes::as_str), Some("-1")); assert!(!retention.read_only); @@ -176,12 +176,6 @@ async fn describe_configs_returns_retention_and_cleanup_and_alter_persists_reten assert_eq!(retention.config_source, SOURCE_DEFAULT); assert_eq!(retention.synonyms, [] as [DescribeConfigsSynonym; 0]); assert!(retention.documentation.is_some()); - let cleanup = entry(&result.configs, "cleanup.policy"); - assert_eq!(cleanup.value.as_ref().map(StrBytes::as_str), Some("delete")); - assert!(cleanup.read_only); - assert_eq!(cleanup.config_source, SOURCE_DEFAULT); - assert_eq!(cleanup.synonyms, [] as [DescribeConfigsSynonym; 0]); - assert!(cleanup.documentation.is_some()); let without_docs = describe( &state, @@ -206,11 +200,11 @@ async fn describe_configs_returns_retention_and_cleanup_and_alter_persists_reten true, ) .await; + // Kafka Connect starts when the entry is missing and refuses a topic that reports `delete`. assert_eq!(filtered.results[0].error_code, ERROR_NONE); - assert_eq!(filtered.results[0].configs.len(), 1); assert_eq!( - filtered.results[0].configs[0].name.as_str(), - "cleanup.policy" + filtered.results[0].configs, + [] as [DescribeConfigsResourceResult; 0] ); let v1 = describe( @@ -222,7 +216,7 @@ async fn describe_configs_returns_retention_and_cleanup_and_alter_persists_reten ) .await; assert_eq!(v1.results[0].error_code, ERROR_NONE); - assert_eq!(v1.results[0].configs.len(), 2); + assert_eq!(v1.results[0].configs.len(), 1); let validated = alter( &state, diff --git a/gateways/kafka/tests/create_topics_real_bridge_tests.rs b/gateways/kafka/tests/create_topics_real_bridge_tests.rs index b2470a55e..8f1f97463 100644 --- a/gateways/kafka/tests/create_topics_real_bridge_tests.rs +++ b/gateways/kafka/tests/create_topics_real_bridge_tests.rs @@ -331,6 +331,51 @@ async fn create_topics_with_a_config_against_an_existing_topic_answers_already_e assert_eq!(error_code, ERROR_TOPIC_ALREADY_EXISTS); } +/// Builds a v3 `CreateTopics` request for one topic with no assignments and no configs. v3 is +/// not flexible, so strings and arrays carry fixed-width lengths. +fn build_v3_request(name: &str, num_partitions: i32, replication_factor: i16) -> Bytes { + let mut enc = Encoder::with_capacity(64); + enc.write_i32(1); // one topic + enc.write_nullable_string(Some(name)) + .expect("topic name fits"); + enc.write_i32(num_partitions); + enc.write_i16(replication_factor); + enc.write_i32(0); // no assignments + enc.write_i32(0); // no configs + enc.write_i32(5_000); // timeout_ms + enc.write_bool(false); // validate_only + enc.freeze() +} + +/// `KafkaJS` `createTopics()` sends `-1` partitions and replication factor at v3. Kafka's +/// controller accepts them at every version, so the topic is created with the default partition +/// count. +#[tokio::test] +#[serial] +async fn create_topics_v3_accepts_broker_default_sentinels() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let state = connected_state(&server).await; + + let outcome = create_topics::handle(&state, 3, build_v3_request("orders", -1, -1)).await; + let mut d = Decoder::new(outcome.expect_response("CreateTopics request always answers")); + let _throttle_time_ms = d.read_i32().expect("throttle_time_ms"); + assert_eq!(d.read_i32().expect("topics array count"), 1); + assert_eq!( + d.read_nullable_string().expect("topic name").as_deref(), + Some("orders") + ); + assert_eq!(d.read_i16().expect("error_code"), ERROR_NONE); + + let raw = raw_client(&server).await; + let topics = raw + .get_topics(&Identifier::named("kafka").expect("valid stream name")) + .await + .expect("get_topics call"); + assert_eq!(topics.len(), 1, "the handler must have actually created it"); + assert_eq!(topics[0].partitions_count, 1); +} + /// A manual replica assignment resolves the created partition count from its own length, not /// from `num_partitions` (`-1` here, per KIP-464). #[tokio::test] diff --git a/gateways/kafka/tests/java/ConfigsProbe.java b/gateways/kafka/tests/java/ConfigsProbe.java index 6c757f404..454c666f2 100644 --- a/gateways/kafka/tests/java/ConfigsProbe.java +++ b/gateways/kafka/tests/java/ConfigsProbe.java @@ -53,26 +53,17 @@ public final class ConfigsProbe { try (AdminClient admin = AdminClient.create(props)) { Config before = describe(admin, resource); ConfigEntry retention = before.get("retention.ms"); - ConfigEntry cleanup = before.get("cleanup.policy"); require(retention != null && "-1".equals(retention.value()), "initial retention.ms"); require( retention.source() == ConfigEntry.ConfigSource.DEFAULT_CONFIG, "initial retention source"); require(!retention.isReadOnly(), "retention.ms is writable"); - require(cleanup != null && "delete".equals(cleanup.value()), "cleanup.policy"); - require(cleanup.isReadOnly(), "cleanup.policy is read only"); - require( - cleanup.source() == ConfigEntry.ConfigSource.DEFAULT_CONFIG, - "cleanup source"); + // Kafka Connect refuses an internal topic that reports any cleanup.policy other + // than compact, and starts when the entry is missing. + require(before.get("cleanup.policy") == null, "cleanup.policy is not reported"); System.out.println( "retention.ms=" + retention.value() + " source=" + retention.source()); - System.out.println( - "cleanup.policy=" - + cleanup.value() - + " read_only=" - + cleanup.isReadOnly() - + " source=" - + cleanup.source()); + System.out.println("cleanup.policy=absent"); try { admin.alterConfigs(Map.of(resource, config("no.such", "1"))).all().get(); diff --git a/gateways/kafka/tests/response_negative_tests.rs b/gateways/kafka/tests/response_negative_tests.rs index 14750559a..02d9cf801 100644 --- a/gateways/kafka/tests/response_negative_tests.rs +++ b/gateways/kafka/tests/response_negative_tests.rs @@ -129,9 +129,10 @@ fn create_topics_v5_flags_invalid_replication_factor() { } #[test] -fn create_topics_v2_rejects_broker_default_sentinel() { - // KIP-464 defaults apply from v4; on v2 without assignments, -1 is INVALID_PARTITIONS. - let req = CreateTopicsRequest::default().with_topics(vec![creatable_topic("legacy", -1, 1)]); +fn create_topics_v2_accepts_broker_default_sentinels() { + // Kafka's controller takes -1 at every version, and KafkaJS sends it at v3. The stub still + // answers NOT_CONTROLLER, not INVALID_PARTITIONS or INVALID_REPLICATION_FACTOR. + let req = CreateTopicsRequest::default().with_topics(vec![creatable_topic("legacy", -1, -1)]); let body = create_topics::encode_response(2, &req).unwrap(); let mut d = Decoder::new(body); assert_eq!(d.read_i32().unwrap(), 0); @@ -140,12 +141,12 @@ fn create_topics_v2_rejects_broker_default_sentinel() { d.read_nullable_string().unwrap(), Some("legacy".to_string()) ); - assert_eq!(d.read_i16().unwrap(), ERROR_INVALID_PARTITIONS); + assert_eq!(d.read_i16().unwrap(), ERROR_NOT_CONTROLLER); } #[test] fn create_topics_v2_with_assignments_allows_broker_default_sentinels() { - // KIP-464: on v2/v3, -1 partitions/replication are valid when assignments are present. + // KIP-464: -1 partitions/replication stay valid on v2 when assignments are present too. use kafka_protocol::messages::create_topics_request::CreatableReplicaAssignment; let assignment = CreatableReplicaAssignment::default()
