This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new b669d9669 fix(gateways): stop reporting cleanup.policy and accept -1
defaults (#4534)
b669d9669 is described below
commit b669d9669b165eed6900f8ed78ddfcdc1f6f3b10
Author: Krishna Vishal <[email protected]>
AuthorDate: Wed Oct 14 14:43:23 2026 +0530
fix(gateways): stop reporting cleanup.policy and accept -1 defaults (#4534)
---
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 b8cd2d962..5a3ef4695 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()
