You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by kr...@apache.org on 2026/10/13 18:19:22 UTC
(iggy) branch kafka-configs-regression created (now 82c050dad)
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a change to branch kafka-configs-regression
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 82c050dad fix(gateways): stop reporting cleanup.policy and accept -1 defaults
This branch includes the following new commits:
new 82c050dad fix(gateways): stop reporting cleanup.policy and accept -1 defaults
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: fix(gateways): stop reporting cleanup.policy and accept -1 defaults
Posted by kr...@apache.org.
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 <kr...@gmail.com>
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()