slbotbm commented on code in PR #4022:
URL: https://github.com/apache/iggy/pull/4022#discussion_r3991298358
##########
foreign/cpp/MODULE.bazel.lock:
##########
@@ -483,109 +483,6 @@
}
}
}
Review Comment:
looks like it was dropped by mistake. restored.
##########
foreign/cpp/tests/e2e/topic.cpp:
##########
@@ -158,171 +155,175 @@ TEST_F(LowLevelE2E_Topic,
CreateTopicWithInvalidNamesThrows) {
};
for (const auto &topic_name : illegal_topic_names) {
SCOPED_TRACE(topic_name);
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", "server_default",
- 0, "server_default", {}),
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(1)),
std::exception);
}
const std::string max_length_name(255, 'a');
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
max_length_name, 1, "none",
- "server_default", 0,
"server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
max_length_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(1)));
}
-TEST_F(LowLevelE2E_Topic, CreateDuplicateTopicThrows) {
+TEST_F(E2E_Topic, CreateDuplicateTopicThrows) {
RecordProperty("description", "Rejects creating a duplicate topic within
the same stream.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", "server_default",
- 0, "server_default", {}));
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", "server_default", 0,
- "server_default", {}),
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(1)),
std::exception);
}
-TEST_F(LowLevelE2E_Topic, CreateSameTopicNameInDifferentStreamsSucceeds) {
+TEST_F(E2E_Topic, CreateSameTopicNameInDifferentStreamsSucceeds) {
RecordProperty("description", "Allows the same topic name to be created in
different streams.");
const std::string first_stream_name = GetRandomName();
const std::string second_stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(first_stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(first_stream_name));
TrackStream(first_stream_name);
- ASSERT_NO_THROW(client->create_stream(second_stream_name));
+ ASSERT_NO_THROW(client.CreateStream(second_stream_name));
TrackStream(second_stream_name);
-
ASSERT_NO_THROW(client->create_topic(make_string_identifier(first_stream_name),
topic_name, 1, "none",
- "server_default", 0,
"server_default", {}));
-
ASSERT_NO_THROW(client->create_topic(make_string_identifier(second_stream_name),
topic_name, 1, "none",
- "server_default", 0,
"server_default", {}));
+
ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(first_stream_name),
topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(1)));
+
ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(second_stream_name),
topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(1)));
}
-TEST_F(LowLevelE2E_Topic, CreateTopicWithInvalidOptionsThrows) {
+TEST_F(E2E_Topic, CreateTopicWithInvalidOptionsThrows) {
RecordProperty("description", "Rejects topic creation requests that use
invalid option values.");
const std::string stream_name = GetRandomName();
const std::string invalid_compression_topic_name = GetRandomName();
const std::string invalid_expiry_topic_name = GetRandomName();
const std::string invalid_max_size_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name),
invalid_compression_topic_name, 1,
- "invalid-compression", "server_default",
0, "server_default", {}),
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
invalid_compression_topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(1).SetRawEntries(
+ {{"compression_algorithm",
"invalid-compression"}})),
std::exception);
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name),
invalid_expiry_topic_name, 1, "none",
- "invalid-expiry-kind", 0,
"server_default", {}),
+ ASSERT_THROW(client.CreateTopic(
+ iggy::Identifier::String(stream_name),
invalid_expiry_topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(1).SetRawEntries({{"message_expiry",
"invalid"}})),
std::exception);
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name),
invalid_max_size_topic_name, 1, "none",
- "server_default", 0, "not-a-size", {}),
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
invalid_max_size_topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(1).SetRawEntries(
+ {{"max_topic_size", "not-a-size"}})),
std::exception);
}
-TEST_F(LowLevelE2E_Topic,
CreateTopicWithOptionsReturnsCanonicalKindAndDerivedRemainder) {
+TEST_F(E2E_Topic,
CreateTopicWithOptionsReturnsCanonicalKindAndDerivedRemainder) {
RecordProperty("description",
"Returns an explicitly set option in its canonical kind,
derives the keys left unset, and rejects "
"an option key outside the server catalog.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string unknown_option_topic = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- rust::Vec<iggy::ffi::HeaderEntry> options;
-
options.push_back(make_header_entry(make_header_field(iggy::ffi::HeaderKind::String,
to_payload("durability")),
-
make_header_field(iggy::ffi::HeaderKind::String, to_payload("persisted"))));
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", "server_default",
- 0, "server_default",
std::move(options)));
+ iggy::TopicCreateOptions options;
+ options.SetPartitionsCount(1).SetDurability(iggy::Durability::Persisted);
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name, options));
const auto topic_details =
- client->get_topic(make_string_identifier(stream_name),
make_string_identifier(topic_name));
+ client.GetTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name));
// Admission re-encodes the block from its own parse, so a value comes back
// in its key's catalog kind rather than in the kind that was sent.
- EXPECT_TRUE(has_header(topic_details.options,
static_cast<std::uint8_t>(iggy::ffi::HeaderKind::String),
- to_payload("durability"),
static_cast<std::uint8_t>(iggy::ffi::HeaderKind::String),
- to_payload("persisted")));
-
- EXPECT_FALSE(topic_details.derived_options.empty());
- std::unordered_set<std::string> derived_option_keys;
- for (const auto &derived_option : topic_details.derived_options) {
-
derived_option_keys.insert(std::string(derived_option.key.value.begin(),
derived_option.key.value.end()));
- }
- EXPECT_EQ(derived_option_keys.count("max_topic_size"), 1u);
- EXPECT_EQ(derived_option_keys.count("durability"), 0u);
-
- rust::Vec<iggy::ffi::HeaderEntry> unknown_options;
- unknown_options.push_back(
- make_header_entry(make_header_field(iggy::ffi::HeaderKind::String,
to_payload("not_a_real_option")),
- make_header_field(iggy::ffi::HeaderKind::String,
to_payload("true"))));
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name),
unknown_option_topic, 1, "none",
- "server_default", 0, "server_default",
std::move(unknown_options)),
+ // The SDK sends both durability fields, including the independently
+ // defaulted offset policy.
+ const auto &explicit_options = topic_details.Options().Explicit();
+ ASSERT_EQ(explicit_options.size(), 2u);
+ auto explicit_it = explicit_options.find("durability");
+ ASSERT_NE(explicit_it, explicit_options.end());
+ EXPECT_EQ(explicit_it->second.Kind(), iggy::HeaderKind::String);
+ EXPECT_EQ(explicit_it->second.Value(), (std::vector<std::uint8_t>{'p',
'e', 'r', 's', 'i', 's', 't', 'e', 'd'}));
+ auto offset_it = explicit_options.find("consumer_offset_durability");
+ ASSERT_NE(offset_it, explicit_options.end());
+ EXPECT_EQ(offset_it->second.Kind(), iggy::HeaderKind::String);
+ EXPECT_EQ(offset_it->second.Value(), (std::vector<std::uint8_t>{'r', 'e',
'p', 'l', 'i', 'c', 'a', 't', 'e', 'd'}));
+
+ EXPECT_FALSE(topic_details.Options().Derived().empty());
+ const auto &derived_options = topic_details.Options().Derived();
+ EXPECT_EQ(derived_options.count("max_topic_size"), 1u);
+ EXPECT_EQ(derived_options.count("durability"), 0u);
+ EXPECT_EQ(derived_options.count("consumer_offset_durability"), 0u);
+
+ iggy::TopicCreateOptions unknown_options;
+ unknown_options.SetPartitionsCount(1).SetRawEntries({{"not_a_real_option",
"true"}});
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
unknown_option_topic, unknown_options),
std::exception);
}
-TEST_F(LowLevelE2E_Topic,
CreateTopicWithTypedOptionHelpersReportsThemAsExplicitOptions) {
+TEST_F(E2E_Topic,
CreateTopicWithTypedOptionHelpersReportsThemAsExplicitOptions) {
RecordProperty("description",
"Creates a topic with every typed option helper and
verifies each key comes back as an explicit "
"option in its catalog kind.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
constexpr std::uint64_t segment_size_bytes = 8ULL * 1024ULL
* 1024ULL;
constexpr std::uint32_t messages_required_to_save = 512;
constexpr std::uint64_t size_of_messages_required_to_save = 2ULL * 1024ULL
* 1024ULL;
- rust::Vec<iggy::ffi::HeaderEntry> options;
- options.push_back(iggy::TopicOption::SegmentSize(segment_size_bytes));
-
options.push_back(iggy::TopicOption::Durability(iggy::Durability::Persisted));
-
options.push_back(iggy::TopicOption::MessagesRequiredToSave(messages_required_to_save));
-
options.push_back(iggy::TopicOption::SizeOfMessagesRequiredToSave(size_of_messages_required_to_save));
- options.push_back(iggy::TopicOption::PreallocateSegments(false));
+ iggy::TopicCreateOptions options;
+ options.SetPartitionsCount(1)
+ .SetSegmentSize(segment_size_bytes)
+ .SetDurability(iggy::Durability::Persisted)
+ .SetMessagesRequiredToSave(messages_required_to_save)
+ .SetSizeOfMessagesRequiredToSave(size_of_messages_required_to_save)
+ .SetPreallocateSegments(false);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", "server_default",
- 0, "server_default",
std::move(options)));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name, options));
const auto topic_details =
- client->get_topic(make_string_identifier(stream_name),
make_string_identifier(topic_name));
-
- const auto key_kind =
static_cast<std::uint8_t>(iggy::ffi::HeaderKind::String);
- const auto bool_kind =
static_cast<std::uint8_t>(iggy::ffi::HeaderKind::Bool);
- const auto uint32_kind =
static_cast<std::uint8_t>(iggy::ffi::HeaderKind::Uint32);
- const auto uint64_kind =
static_cast<std::uint8_t>(iggy::ffi::HeaderKind::Uint64);
-
- EXPECT_TRUE(has_header(topic_details.options, key_kind,
to_payload("segment_size"), uint64_kind,
- little_endian_bytes(segment_size_bytes, 8)));
- EXPECT_TRUE(
- has_header(topic_details.options, key_kind, to_payload("durability"),
key_kind, to_payload("persisted")));
- EXPECT_TRUE(has_header(topic_details.options, key_kind,
to_payload("messages_required_to_save"), uint32_kind,
- little_endian_bytes(messages_required_to_save, 4)));
- EXPECT_TRUE(has_header(topic_details.options, key_kind,
to_payload("size_of_messages_required_to_save"),
- uint64_kind,
little_endian_bytes(size_of_messages_required_to_save, 8)));
- EXPECT_TRUE(
- has_header(topic_details.options, key_kind,
to_payload("preallocate_segments"), bool_kind, bool_bytes(false)));
-
- for (const auto &derived_option : topic_details.derived_options) {
- const std::string derived_key(derived_option.key.value.begin(),
derived_option.key.value.end());
- EXPECT_NE(derived_key, "segment_size") << "segment_size was set
explicitly, so it cannot be derived";
- }
+ client.GetTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name));
+
+ const auto &explicit_options = topic_details.Options().Explicit();
+ EXPECT_EQ(explicit_options.count("segment_size"), 1u);
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
