This is an automated email from the ASF dual-hosted git repository. spetz pushed a commit to branch resource_options in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 5cceaa9f71ff6ff912fa33b6e7f3d9093c154784 Author: spetz <[email protected]> AuthorDate: Wed Oct 21 15:58:49 2026 +0200 fix --- .gitignore | 1 + .../__pycache__/conftest.cpython-311-pytest-9.1.1.pyc | Bin 4859 -> 0 bytes ...resource_options_patch.cpython-311-pytest-9.1.1.pyc | Bin 27435 -> 0 bytes .../test_stream_options.cpython-311-pytest-9.1.1.pyc | Bin 42296 -> 0 bytes gateways/kafka/src/bridge/iggy_bridge/cluster_id.rs | 12 +++++++----- gateways/kafka/tests/cluster_id_real_bridge_tests.rs | 17 +++++++++++------ 6 files changed, 19 insertions(+), 11 deletions(-) diff --git a/.gitignore b/.gitignore index e5e5ce683..2dfad0276 100644 --- a/.gitignore +++ b/.gitignore @@ -49,6 +49,7 @@ foreign/node/.npm *.lib *.out *.pid +__pycache__/ go.work core/bench/dashboard/frontend/dist core/integration/tests/connectors/postgres/cdc_restart_connectors/* diff --git a/bdd/python/tests/__pycache__/conftest.cpython-311-pytest-9.1.1.pyc b/bdd/python/tests/__pycache__/conftest.cpython-311-pytest-9.1.1.pyc deleted file mode 100644 index 7a5aeedae..000000000 Binary files a/bdd/python/tests/__pycache__/conftest.cpython-311-pytest-9.1.1.pyc and /dev/null differ diff --git a/bdd/python/tests/__pycache__/test_resource_options_patch.cpython-311-pytest-9.1.1.pyc b/bdd/python/tests/__pycache__/test_resource_options_patch.cpython-311-pytest-9.1.1.pyc deleted file mode 100644 index e506431c9..000000000 Binary files a/bdd/python/tests/__pycache__/test_resource_options_patch.cpython-311-pytest-9.1.1.pyc and /dev/null differ diff --git a/bdd/python/tests/__pycache__/test_stream_options.cpython-311-pytest-9.1.1.pyc b/bdd/python/tests/__pycache__/test_stream_options.cpython-311-pytest-9.1.1.pyc deleted file mode 100644 index 92272a6fa..000000000 Binary files a/bdd/python/tests/__pycache__/test_stream_options.cpython-311-pytest-9.1.1.pyc and /dev/null differ diff --git a/gateways/kafka/src/bridge/iggy_bridge/cluster_id.rs b/gateways/kafka/src/bridge/iggy_bridge/cluster_id.rs index 141e47908..c69b49d48 100644 --- a/gateways/kafka/src/bridge/iggy_bridge/cluster_id.rs +++ b/gateways/kafka/src/bridge/iggy_bridge/cluster_id.rs @@ -26,8 +26,8 @@ use std::fmt; use iggy::prelude::{ Consumer, Durability, Identifier, IggyError, IggyExpiry, IggyMessage, MaxTopicSize, - MessageClient, Partitioning, PollingStrategy, StreamClient, TopicClient, TopicCreateOptions, - TopicDetails, + MessageClient, Partitioning, PollingStrategy, StreamClient, StreamCreateOptions, TopicClient, + TopicCreateOptions, TopicDetails, }; use tracing::{info, warn}; @@ -203,8 +203,9 @@ impl IggyBridge { } async fn create_cluster_id_stream(&self, store: &IdStore) -> Result<(), BridgeError> { + let options = StreamCreateOptions::default(); match self - .metadata_write(|| self.client.create_stream(&store.stream_name)) + .metadata_write(|| self.client.create_stream(&store.stream_name, &options)) .await { Ok(_) | Err(BridgeError::Iggy(IggyError::StreamNameAlreadyExists(_))) => Ok(()), @@ -314,11 +315,12 @@ impl IdTopicOptions { impl From<&TopicDetails> for IdTopicOptions { fn from(topic: &TopicDetails) -> Self { + let resolved = topic.resolved_options(); Self { partitions_count: topic.partitions_count, durability: TopicCreateOptions::from_resource_options(&topic.options).durability, - message_expiry: topic.message_expiry, - max_topic_size: topic.max_topic_size, + message_expiry: resolved.message_expiry, + max_topic_size: resolved.max_topic_size, } } } diff --git a/gateways/kafka/tests/cluster_id_real_bridge_tests.rs b/gateways/kafka/tests/cluster_id_real_bridge_tests.rs index 3a5cd76a1..0083d193f 100644 --- a/gateways/kafka/tests/cluster_id_real_bridge_tests.rs +++ b/gateways/kafka/tests/cluster_id_real_bridge_tests.rs @@ -25,8 +25,8 @@ use bytes::Bytes; use futures::future::join_all; use iggy::prelude::{ Consumer, Durability, Identifier, IggyExpiry, IggyMessage, MaxTopicSize, MessageClient, - Partitioning, Permissions, PollingStrategy, StreamClient, StreamPermissions, TopicClient, - TopicCreateOptions, UserClient, UserStatus, + Partitioning, Permissions, PollingStrategy, StreamClient, StreamCreateOptions, + StreamPermissions, TopicClient, TopicCreateOptions, UserClient, UserCreateOptions, UserStatus, }; use kafka_protocol::messages::{DescribeClusterResponse, MetadataResponse}; use kafka_protocol::protocol::Decodable; @@ -70,7 +70,10 @@ async fn connect_scoped( ) -> IggyBridge { let raw = raw_client(server).await; let stream = raw - .create_stream(&cluster_id_stream(DEFAULT_STREAM)) + .create_stream( + &cluster_id_stream(DEFAULT_STREAM), + &StreamCreateOptions::default(), + ) .await .expect("create stream"); raw.create_user( @@ -81,6 +84,7 @@ async fn connect_scoped( streams: Some(BTreeMap::from([(stream.id as usize, permissions)])), ..Permissions::default() }), + &UserCreateOptions::default(), ) .await .expect("create user"); @@ -179,8 +183,9 @@ async fn given_a_new_id_when_stored_should_be_persisted_never_expire_and_have_no .expect("get_topic") .expect("the id topic exists"); assert_eq!(topic.partitions_count, 1); - assert_eq!(topic.message_expiry, IggyExpiry::NeverExpire); - assert_eq!(topic.max_topic_size, MaxTopicSize::Unlimited); + let resolved = topic.resolved_options(); + assert_eq!(resolved.message_expiry, IggyExpiry::NeverExpire); + assert_eq!(resolved.max_topic_size, MaxTopicSize::Unlimited); assert_eq!( TopicCreateOptions::from_resource_options(&topic.options).durability, Durability::Persisted @@ -244,7 +249,7 @@ async fn given_a_stored_message_that_is_not_an_id_when_loaded_should_refuse_it() let server = TestServer::spawn(data_dir.path()).await; let raw = raw_client(&server).await; let stream_name = cluster_id_stream(DEFAULT_STREAM); - raw.create_stream(&stream_name) + raw.create_stream(&stream_name, &StreamCreateOptions::default()) .await .expect("create stream"); let stream_id = Identifier::named(&stream_name).expect("stream name");
