feat(common)!: extensible key-value options for topics, streams, users

Cargo.toml

@@ -227,7 +227,6 @@ metadata = { path = "core/metadata" }
 mimalloc = "0.1"
 mime_guess = "2.0"
 mockall = "0.15.0"
-moka = { version = "0.12.15", features = ["future"] }
 mongodb = { version = "3.8.0", features = ["rustls-tls"] }
 nix = { version = "0.31.3", features = ["feature", "fs", "resource", "sched"] }
 nonzero_lit = "0.1.2"

bdd/cpp/features/step_definitions/messaging_steps.cpp

@@ -78,9 +78,9 @@ WHEN("^I create a topic with name \"([^\"]{1,255})\" in stream ([0-9]+) with ([0
 
     context->client->create_topic(bdd::make_numeric_identifier(static_cast<std::uint32_t>(stream_id)), topic_name,
                                   static_cast<std::uint32_t>(partitions_count),
-                                  std::string(compression.compression_algorithm_value()), 0,
+                                  std::string(compression.compression_algorithm_value()),
                                   std::string(message_expiry.expiry_kind()), message_expiry.expiry_value(),
-                                  std::string(max_topic_size.max_topic_size()));
+                                  std::string(max_topic_size.max_topic_size()), {});
 }
 
 THEN("^the topic should be created successfully$") {
@@ -117,9 +117,9 @@ WHEN("^I send ([0-9]+) messages to stream ([0-9]+), topic ([0-9]+), partition ([
 
     rust::Vec<iggy::ffi::IggyMessageToSend> messages;
     for (int index = 0; index < message_count; ++index) {
-        iggy::ffi::IggyMessageToSend message = iggy::ffi::make_message(
-            bdd::to_payload(bdd::expected_payload(static_cast<std::uint32_t>(index))),
-            rust::Vec<iggy::ffi::HeaderEntry>());
+        iggy::ffi::IggyMessageToSend message =
+            iggy::ffi::make_message(bdd::to_payload(bdd::expected_payload(static_cast<std::uint32_t>(index))),
+                                    rust::Vec<iggy::ffi::HeaderEntry>());
         // Assign an explicit, 1-based id so the last-sent/last-polled comparison is meaningful.
         message.id_lo = static_cast<std::uint64_t>(index + 1);
         messages.push_back(std::move(message));

bdd/go/tests/basic_messaging.go

@@ -269,7 +269,6 @@ func (s basicMessagingSteps) whenCreateTopic(ctx context.Context,
 		iggcon.CompressionAlgorithmNone,
 		iggcon.IggyExpiryNeverExpire,
 		0,
-		nil,
 	)
 	if err != nil {
 		return fmt.Errorf("failed to create topic: %w", err)

bdd/go/tests/tcp_test/stream_feature_get_by_id.go

@@ -63,8 +63,7 @@ var _ = ginkgo.Describe("GET STREAM BY ID:", func() {
 				2,
 				iggcon.CompressionAlgorithmNone,
 				iggcon.Millisecond,
-				math.MaxUint64,
-				nil)
+				math.MaxUint64)
 			itShouldNotReturnError(err)
 			t2, err := client.CreateTopic(
 				context.Background(),
@@ -73,8 +72,7 @@ var _ = ginkgo.Describe("GET STREAM BY ID:", func() {
 				2,
 				iggcon.CompressionAlgorithmNone,
 				iggcon.Millisecond,
-				math.MaxUint64,
-				nil)
+				math.MaxUint64)
 			itShouldNotReturnError(err)
 			itShouldSuccessfullyCreateTopic(streamId, t1.Id, t1Name, client)
 			itShouldSuccessfullyCreateTopic(streamId, t2.Id, t2Name, client)

bdd/go/tests/tcp_test/topic_feature_create.go

@@ -32,7 +32,6 @@ var _ = ginkgo.Describe("CREATE TOPIC:", func() {
 		ginkgo.Context("and tries to create topic unique name and id", func() {
 			client := createAuthorizedConnection()
 			streamId, _ := successfullyCreateStream(prefix, client)
-			replicationFactor := uint8(1)
 			name := createRandomString(32)
 			defer deleteStreamAfterTests(streamId, client)
 			streamIdentifier, _ := iggcon.NewIdentifier(streamId)
@@ -43,16 +42,14 @@ var _ = ginkgo.Describe("CREATE TOPIC:", func() {
 				2,
 				iggcon.CompressionAlgorithmNone,
 				iggcon.Millisecond,
-				math.MaxUint64,
-				&replicationFactor)
+				math.MaxUint64)
 
 			itShouldNotReturnError(err)
 		})
 
 		ginkgo.Context("and tries to create topic for a non existing stream", func() {
 			client := createAuthorizedConnection()
 			streamId := createRandomUInt32()
-			replicationFactor := uint8(1)
 			name := createRandomString(32)
 			streamIdentifier, _ := iggcon.NewIdentifier(streamId)
 			_, err := client.CreateTopic(
@@ -62,8 +59,7 @@ var _ = ginkgo.Describe("CREATE TOPIC:", func() {
 				2,
 				iggcon.CompressionAlgorithmNone,
 				iggcon.Millisecond,
-				math.MaxUint64,
-				&replicationFactor)
+				math.MaxUint64)
 
 			itShouldReturnSpecificError(err, ierror.ErrStreamIdNotFound)
 		})
@@ -74,7 +70,6 @@ var _ = ginkgo.Describe("CREATE TOPIC:", func() {
 			defer deleteStreamAfterTests(streamId, client)
 			_, name := successfullyCreateTopic(streamId, client)
 
-			replicationFactor := uint8(1)
 			streamIdentifier, _ := iggcon.NewIdentifier(streamId)
 			_, err := client.CreateTopic(
 				context.Background(),
@@ -83,8 +78,7 @@ var _ = ginkgo.Describe("CREATE TOPIC:", func() {
 				2,
 				iggcon.CompressionAlgorithmNone,
 				iggcon.IggyExpiryServerDefault,
-				math.MaxUint64,
-				&replicationFactor)
+				math.MaxUint64)
 			itShouldReturnSpecificError(err, ierror.ErrTopicNameAlreadyExists)
 		})
 
@@ -94,16 +88,14 @@ var _ = ginkgo.Describe("CREATE TOPIC:", func() {
 			defer deleteStreamAfterTests(streamId, createAuthorizedConnection())
 
 			streamIdentifier, _ := iggcon.NewIdentifier(streamId)
-			replicationFactor := uint8(1)
 			_, err := client.CreateTopic(
 				context.Background(),
 				streamIdentifier,
 				createRandomString(256),
 				2,
 				iggcon.CompressionAlgorithmNone,
 				iggcon.IggyExpiryServerDefault,
-				math.MaxUint64,
-				&replicationFactor)
+				math.MaxUint64)
 
 			itShouldReturnSpecificError(err, ierror.ErrInvalidTopicName)
 		})
@@ -112,7 +104,6 @@ var _ = ginkgo.Describe("CREATE TOPIC:", func() {
 	ginkgo.When("User is not logged in", func() {
 		ginkgo.Context("and tries to create topic", func() {
 			client := createClient()
-			replicationFactor := uint8(1)
 			streamIdentifier, _ := iggcon.NewIdentifier[uint32](10)
 			_, err := client.CreateTopic(
 				context.Background(),
@@ -121,8 +112,7 @@ var _ = ginkgo.Describe("CREATE TOPIC:", func() {
 				2,
 				iggcon.CompressionAlgorithmNone,
 				iggcon.IggyExpiryServerDefault,
-				math.MaxUint64,
-				&replicationFactor)
+				math.MaxUint64)
 
 			itShouldReturnUnauthenticatedError(err)
 		})

bdd/go/tests/tcp_test/topic_feature_update.go

@@ -35,7 +35,6 @@ var _ = ginkgo.Describe("UPDATE TOPIC:", func() {
 			defer deleteStreamAfterTests(streamId, client)
 			topicId, _ := successfullyCreateTopic(streamId, client)
 			newName := createRandomString(128)
-			replicationFactor := uint8(1)
 			streamIdentifier, _ := iggcon.NewIdentifier(streamId)
 			topicIdentifier, _ := iggcon.NewIdentifier(topicId)
 			err := client.UpdateTopic(
@@ -45,8 +44,7 @@ var _ = ginkgo.Describe("UPDATE TOPIC:", func() {
 				newName,
 				iggcon.CompressionAlgorithmNone,
 				iggcon.Microsecond,
-				math.MaxUint64,
-				&replicationFactor)
+				math.MaxUint64)
 			itShouldNotReturnError(err)
 			itShouldSuccessfullyUpdateTopic(streamId, topicId, newName, client)
 		})
@@ -57,7 +55,6 @@ var _ = ginkgo.Describe("UPDATE TOPIC:", func() {
 			defer deleteStreamAfterTests(streamId, client)
 			_, topic1Name := successfullyCreateTopic(streamId, client)
 			topic2Id, _ := successfullyCreateTopic(streamId, client)
-			replicationFactor := uint8(1)
 			streamIdentifier, _ := iggcon.NewIdentifier(streamId)
 			topic2Identifier, _ := iggcon.NewIdentifier(topic2Id)
 			err := client.UpdateTopic(
@@ -67,24 +64,21 @@ var _ = ginkgo.Describe("UPDATE TOPIC:", func() {
 				topic1Name,
 				iggcon.CompressionAlgorithmNone,
 				iggcon.IggyExpiryServerDefault,
-				math.MaxUint64,
-				&replicationFactor)
+				math.MaxUint64)
 
 			itShouldReturnSpecificError(err, ierror.ErrTopicNameAlreadyExists)
 		})
 
 		ginkgo.Context("and tries to update non-existing topic", func() {
 			client := createAuthorizedConnection()
-			replicationFactor := uint8(1)
 			err := client.UpdateTopic(
 				context.Background(),
 				randomU32Identifier(),
 				randomU32Identifier(),
 				createRandomString(128),
 				1,
 				0,
-				math.MaxUint64,
-				&replicationFactor)
+				math.MaxUint64)
 
 			itShouldReturnSpecificError(err, ierror.ErrStreamIdNotFound)
 		})
@@ -93,7 +87,6 @@ var _ = ginkgo.Describe("UPDATE TOPIC:", func() {
 			client := createAuthorizedConnection()
 			streamId, _ := successfullyCreateStream(prefix, client)
 			defer deleteStreamAfterTests(streamId, createAuthorizedConnection())
-			replicationFactor := uint8(1)
 			streamIdentifier, _ := iggcon.NewIdentifier(streamId)
 			err := client.UpdateTopic(
 				context.Background(),
@@ -102,8 +95,7 @@ var _ = ginkgo.Describe("UPDATE TOPIC:", func() {
 				createRandomString(128),
 				1,
 				0,
-				math.MaxUint64,
-				&replicationFactor)
+				math.MaxUint64)
 
 			itShouldReturnSpecificError(err, ierror.ErrTopicIdNotFound)
 		})
@@ -113,7 +105,6 @@ var _ = ginkgo.Describe("UPDATE TOPIC:", func() {
 			streamId, _ := successfullyCreateStream(prefix, client)
 			defer deleteStreamAfterTests(streamId, createAuthorizedConnection())
 			topicId, _ := successfullyCreateTopic(streamId, client)
-			replicationFactor := uint8(1)
 			streamIdentifier, _ := iggcon.NewIdentifier(streamId)
 			topicIdentifier, _ := iggcon.NewIdentifier(topicId)
 			err := client.UpdateTopic(
@@ -123,8 +114,7 @@ var _ = ginkgo.Describe("UPDATE TOPIC:", func() {
 				createRandomString(256),
 				1,
 				0,
-				math.MaxUint64,
-				&replicationFactor)
+				math.MaxUint64)
 
 			itShouldReturnSpecificError(err, ierror.ErrInvalidTopicName)
 		})

bdd/go/tests/tcp_test/topic_steps.go

@@ -31,7 +31,6 @@ import (
 //operations
 
 func successfullyCreateTopic(streamId uint32, client iggcon.Client) (uint32, string) {
-	replicationFactor := uint8(1)
 	name := createRandomString(128)
 	streamIdentifier, _ := iggcon.NewIdentifier(streamId)
 	topic, err := client.CreateTopic(
@@ -41,8 +40,7 @@ func successfullyCreateTopic(streamId uint32, client iggcon.Client) (uint32, str
 		2,
 		1,
 		0,
-		math.MaxUint64,
-		&replicationFactor)
+		math.MaxUint64)
 
 	itShouldSuccessfullyCreateTopic(streamId, topic.Id, name, client)
 	itShouldNotReturnError(err)

bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java

@@ -108,7 +108,6 @@ public void createTopic(String topicName, int streamId, int partitions) {
                         CompressionAlgorithm.None,
                         BigInteger.ZERO,
                         BigInteger.ZERO,
-                        Optional.empty(),
                         topicName);
 
         context.lastTopicId = topic.id();

bdd/rust/tests/steps/topics.rs

@@ -17,7 +17,7 @@
 
 use crate::common::global_context::GlobalContext;
 use cucumber::{then, when};
-use iggy::prelude::{CompressionAlgorithm, Identifier, IggyExpiry, MaxTopicSize, TopicClient};
+use iggy::prelude::{Identifier, IggyExpiry, TopicClient, TopicCreateOptions};
 
 #[when(regex = r"^I create a topic with name (.+) in stream (\d+) with (\d+) partitions$")]
 pub async fn when_create_topic(
@@ -31,11 +31,11 @@ pub async fn when_create_topic(
         .create_topic(
             &Identifier::numeric(stream_id).unwrap(),
             &topic_name,
-            partitions_count,
-            CompressionAlgorithm::default(),
-            None,
-            IggyExpiry::NeverExpire,
-            MaxTopicSize::ServerDefault,
+            &TopicCreateOptions {
+                partitions_count: Some(partitions_count),
+                message_expiry: Some(IggyExpiry::NeverExpire),
+                ..TopicCreateOptions::default()
+            },
         )
         .await
         .expect("Should be able to create topic");