feat(cpp): add messaging FFI functions for C++ SDK by seokjin0414 · Pull Request #3046 · apache/iggy

seokjin0414 · 2026-03-29T09:43:52Z

Summary

Closes #2965 (Phase 1 of 2).

Add get_streams, send_messages, poll_messages FFI functions to C++ SDK
Add shared structs: Stream, IggyMessageToSend, IggyMessagePolled, PolledMessages
Add 6 e2e tests for message round-trip
All additions are purely additive — no existing FFI changes

Discussed with @slbotbm on Discord — PR split and message struct design confirmed.

Test plan

cargo check / cargo clippy / cargo fmt clean
6 e2e tests in tests/message/low_level_e2e.cpp (Bazel + iggy-server)

slbotbm · 2026-03-29T10:52:19Z

@seokjin0414 Thanks for the PR!

Here is my first round of review:
The CI is failing. Please check which test is failing. Also, the cpp code is likely not linted properly, so please do that as well.

Let's unify the Message struct, instead of having separate structs for sending and polling. The unified struct could look like:

struct Message {
      checksum: u64,
      id_lo: u64,
      id_hi: u64,
      offset: u64,
      timestamp: u64,
      origin_timestamp: u64,
      user_headers_length: u32,
      payload_length: u32,
      reserved: u64,
      payload: Vec<u8>,
      user_headers: Vec<u8>,
  }

Define a function using impl Message that would allow the user to create a message to send. This has to be exported to the cpp side. If the bdd tests only require the payload to be set, you can just define a function called from(payload: Vec<u8>).

You are using map(Into::into) in many places. Let's use from to be more explicit about what is being converted to what.

let mut iggy_messages: Vec<IggyMessage> = messages
            .into_iter()
            .map(|m| {
                let id = ((m.id_hi as u128) << 64) | (m.id_lo as u128);
                let payload = Bytes::from(m.payload);
                let msg = if id > 0 {
                    IggyMessage::builder().id(id).payload(payload).build()
                } else {
                    IggyMessage::builder().payload(payload).build()
                };
                msg.map_err(|error| format!("Could not build message: {error}"))
            })
            .collect::<Result<Vec<_>, _>>()?;

In the above code, please define TryFrom<ffi::Message> for IggyMessage for the conversion and use that.

Let's rename strategy_kind and strategy_value to partitioning_strategy_kind and partitioning_strategy_value respectively so that there is no ambiguity about its meaning.

seokjin0414 · 2026-03-29T11:37:20Z

@slbotbm Thanks for the review! Addressed all feedback in c8f9adf:

CI fix: clang-format applied, fragile s.id > 0 assertion removed (75c2a28)
Unified Message struct: Merged IggyMessageToSend/IggyMessagePolled into single Message with all header fields. Added new_message(payload) factory function exported to C++
TryFrom: Defined TryFrom<ffi::Message> for IggyMessage and use it in send_messages
Explicit from: Replaced map(Into::into) with map(ffi::Stream::from) / map(ffi::Message::from)
Naming: Renamed to polling_strategy_kind/polling_strategy_value — these refer to the polling strategy (offset/timestamp/first/last/next), not partitioning. Partitioning params (partitioning_kind/partitioning_value) are on send_messages separately

slbotbm · 2026-03-29T15:47:42Z

@seokjin0414 Thanks for the quick turn-around.

I may not have been clear in my review previously, but I wanted you to implement something more like

impl ffi::Message {
      pub fn new_message(&mut self, payload: Vec<u8>) -> Result<(), String> {
          if payload.is_empty() {
              return Err("Could not create message: payload must not be empty".to_string());
          }

          let payload_length = payload.len() as u32;

          *self = Self {
              checksum: 0,
              id_lo: 0,
              id_hi: 0,
              offset: 0,
              timestamp: 0,
              origin_timestamp: 0,
              user_headers_length: 0,
              payload_length,
              reserved: 0,
              payload,
              user_headers: Vec::new(),
          };

          Ok(())
      }
}

The above interface would allow us to call Message.new_message(...) from the cpp side while avoiding ambiguity.

GetStreamsReturnsEmptyAfterCleanup and GetStreamsReturnsStreamAfterCreation test the get_streams function, but they are in the messages/ folder. Please move them to their appropriate folders.

The amount of testing is insufficient for the three functions.
Please think about the following and add tests:

get_streams:
- For GetStreamsReturnsStreamAfterCreation, verify everything that is verifiable, not just the name. Also create additional streams and topics inside them, and verify everything.
- Does the sdk allow calling get_streams before login and connect?
- Does calling get_stream and get_streams when there is only one one stream return the same stream with the same details?
- What happens when you call get_streams repeatedly?
- Lastly, in GetStreamsReturnsEmptyAfterCleanup, you are deleting every stream without ever creating any. Instead of this, let's create some unique streams, delete those, and then verify whether 0 is returned or not. Ignore this please.
send_messages:
- What happens when you call send_messages before login and connect?
- What happens when you give invalid topic and / or stream identifiers to the function?
- What happens when you give non-existent topic and / or stream identifiers to the function?
- What happens when you give wrong partitioning kind and / or value to the function? (also test the relevant limits)
- Verify that the messages are being stored in partitions according to the strategy we have strategy we have chosen.
- What happens when you supply an empty vector to the send_messages?
- Is there a limit on how many messages can be sent in total, or in a single batch?
- What happens when you try to send an invalid payload through send_messages (empty payload, payload size > 64,000,000 bytes)
- When send_messages is called multiple times, is the order of the messages preserved?
- You have written a test where one custom ID is set and sent to the server. What happens when you set the same ID for multiple messages?
- Lastly, since we are working with bytes, I would like to you to try sending a variety of payloads, and check whether they succeed or not. (null bytes, utf-8 strings, emojis, etc. - consider using property based testing)
poll_messages :
- What happens when you call poll_messages before login and connect?
- What happens when you give invalid topic, stream, and / or consumer identifiers to the function?
- What happens when you give non-existent topic, stream, and / or consumer identifiers to the function?
- What happens when you give a wrong consumer_kind to the function?
- What happens when you give a wrong polling strategy kind and / or value to the function? (also test the relevant limits)
- What happens when you give an invalid or non-existent partition ID to the function?
- What happens when you supply count = 0 to the poll_messages?
- Is there a limit on how many messages can be requested in a single poll?
- When the requested count is smaller than the number of available messages, are only that many messages returned?
- When the requested count is larger than the number of available messages, are only the available messages returned?
- Verify that polling without specifying a partition behaves as expected.
- Verify the behavior of each polling strategy: offset, first, last, next, and timestamp.
- For offset, what happens when the offset is valid, too large, or points exactly to the end?
- For timestamp, what happens when the timestamp is before the first message, between messages, and after the last message?
- What happens when auto_commit = false and poll_messages is called repeatedly with next?
- What happens when auto_commit = true and poll_messages is called repeatedly with next?
- Are offsets tracked independently for different consumer IDs?
- Are offsets tracked independently for consumer and consumer_group?
- When multiple consumers in the same consumer group poll, is the behavior what we expect?
- When poll_messages is called multiple times after multiple send_messages calls, is the order of the returned messages preserved?
- Verify that the returned offsets are monotonic and correct across multiple polls.
- Verify that the returned payloads are preserved exactly for a variety of byte payloads. (null bytes, utf-8 strings, emojis, arbitrary bytes, etc. - consider doing property-based testing)
- Verify that custom message IDs, if present, are returned correctly for multiple messages, not just one.
- What happens when you poll from an empty partition, and then poll again after new messages are sent?
- What happens when you poll after the stream or topic has been deleted?

slbotbm · 2026-03-29T15:55:44Z

I wrote the above in multiple sittings, so please tell me if I wrote overlapping things.

hubcio · 2026-03-30T07:01:05Z

One comment from my side: remember that Rust SDK (which you are calling) is also perform these validations and returns respective errors. Make sure that you are not double checking this on rust wrapper side or cpp side.

seokjin0414 · 2026-03-30T16:32:44Z

@slbotbm

new_message changed to impl Message method, payload validation removed
get_streams tests moved to tests/stream/
Added 30 new e2e tests (4 get_streams, 11 send_messages, 15 poll_messages) — consumer group tests excluded pending feat(cpp): Add functions related to consumer groups #2988

slbotbm · 2026-03-30T17:34:07Z

@seokjin0414 Thanks for the update. Could you please tell me which tests you have not added from the above list? Just copy pasting the questions I had written will do.

slbotbm · 2026-03-30T17:36:37Z

@hubcio about your comment: would it make sense to add some of these test to the rust sdk instead? ~~I ask because some of the e2e tests can become unit tests on the rust side.~~ I ask because we would also not have to duplicate these tests for the python sdk

seokjin0414 · 2026-03-31T15:33:27Z

@slbotbm
Tests not added:

send_messages:

Is there a limit on how many messages can be sent in total, or in a single batch? — need to investigate the actual server limit first
Property-based testing — used manual cases (null bytes, UTF-8, binary) instead

poll_messages:

Invalid or non-existent partition ID — overlooked, will add
count = 0 — overlooked, will add
Limit on how many messages can be requested in a single poll — need to investigate the actual server limit first
Polling without specifying a partition — overlooked, will add
Timestamp strategy (before/between/after) — overlooked, will add
Monotonic offset verification across multiple polls — overlooked, will add
Consumer vs consumer_group offset independence — pending feat(cpp): Add functions related to consumer groups #2988
Multiple consumers in same consumer group — pending feat(cpp): Add functions related to consumer groups #2988
Property-based testing — used manual cases instead

I can add the overlooked ones now.

slbotbm · 2026-03-31T16:12:18Z

@seokjin0414 please go ahead and add the ones you overlooked. I'll review then

seokjin0414 · 2026-04-01T00:39:01Z

@slbotbm
Done. Added the overlooked tests:

Your question	Test added
Invalid or non-existent partition ID	`PollMessagesWithInvalidPartitionId` — partition 9999 on topic with 1 partition → throws
count = 0	`PollMessagesWithCountZero` — returns success with 0 messages (server returns empty, not error)
Polling without specifying a partition	`PollMessagesWithoutSpecifyingPartition` — UINT32_MAX defaults to partition 0
Timestamp strategy (before/between/after)	`PollMessagesTimestampStrategy` — 2 batches with 100ms gap, verifies >= semantics
Monotonic offset verification	`PollMessagesMonotonicOffsets` — 20 messages, 4 polls of 5, verifies continuous offsets 0-19
Batch size limit	`SendMessagesLargeBatch` — 1000 messages in single batch, all retrievable
Poll request limit	`PollMessagesLargeCount` — count=UINT32_MAX, returns only available 10 messages

Still not added:

Consumer group tests — pending feat(cpp): Add functions related to consumer groups #2988
Property-based testing — covered with manual cases (null bytes, UTF-8, binary)

slbotbm · 2026-04-01T09:03:01Z

@seokjin0414 Thanks for the update! I'll review this by friday/saturday. In the meantime, could you please:

rebase this pr and fix conflicts, since feat(cpp): Add functions related to consumer groups #2988 was recently merged
Fix the failing CI
write unit tests for Message.new_message in tests/messages/unit_tests.cpp? You could refer to https://github.com/apache/iggy/blob/master/foreign/cpp/tests/identifier/unit_tests.cpp to think about what kind of unit tests to write.

seokjin0414 · 2026-04-04T16:19:56Z

@slbotbm

Renamed 10 tests with Throws suffix
Added payload/offset verification:
- PollMessagesFirstStrategy — payload + offset for all 3 messages
- PollMessagesLastStrategy — payload for messages at indices 7-9
- PollMessagesNextStrategyNoAutoCommit — payload + offset for both polls
- PollMessagesNextStrategyAutoCommit — payload for polled1 (0-4) and polled2 (5-9)
- PollMessagesMultipleSendsThenPollOrder — offset for all 10 messages
- PollMessagesWithoutSpecifyingPartition — payload for 5 messages
- PollMessagesTimestampStrategy — payload for batch2 messages
Deleted PollMessagesLargeCount
Removed accidental #[allow] from identifier.rs

slbotbm

@seokjin0414 I requested some small changes.
Other than this, I noticed

you were not testing cases for send_messages and poll_messages when invalid topic ID was passed to them.
No test for invalid consumer identifier for poll_messages

Also, since the consumer group PR was merged, could you add a test for creating a consumer group and polling messages after joining that?

seokjin0414

Thank you for the review!

Added field assertions to GetStreamsReturnsStreamAfterCreation (created_at, size_bytes, messages_count, topics_count) and field comparison to GetStreamsConsistentWithGetStream (created_at, size_bytes). Note: id is excluded from both due to a known cxx Vec<SharedStruct> issue where id returns 0.
Added SendMessagesWithInvalidTopicIdThrows, PollMessagesWithInvalidTopicIdThrows, PollMessagesWithInvalidConsumerIdThrows
Added join_consumer_group and leave_consumer_group FFI functions
Added ConsumerGroupCreateJoinAndPollMessages test covering the full flow: create → join → verify member count → send → poll with consumer_group kind → verify payloads → leave → verify member count drops to 0

slbotbm

@seokjin0414 Thanks for the update!

Could you add the following comment in tests/client/low_level_e2e.cpp please:

TODO(slbotbm): Add tests for join_consumer_group() and leave_consumer_group()

Also, you commented:

Note: id is excluded from both due to a known cxx Vec issue where id returns 0.

Do you have an issue you could link to for me to see?

Otherwise lgtm!

seokjin0414 · 2026-04-08T14:55:58Z

@slbotbm
I investigated further it's not a cxx bug
Iggy uses a slab allocator for stream IDs (metadata/reader.rs:202, vacant_key()) so the first stream gets id=0
The original ASSERT_GT(s.id, 0u) was simply wrong for that reason
I've added EXPECT_EQ(list_id, single.id) to GetStreamsConsistentWithGetStream instead which compares the two API results without assuming a specific value

amlel-el-mahrouss

LGTM, Nothing to add.

slbotbm · 2026-04-12T08:11:31Z

Part of #2100

hubcio

besides below review I noticed:

foreign/cpp/src/client.rs:33-54 new_connection returns *mut Client via Box::into_raw. ownership contract is undocumented - who owns it, who calls delete_connection, what happens on double-free. add a doc-comment.
foreign/cpp/src/client.rs:586 the TODO about logout_user failing leaving a leak is misleading. Box::from_raw/drop runs unconditionally before the error is classified, so there's no leak path. either remove the TODO or rephrase what it actually warns about.
foreign/cpp/src/lib.rs:29 enable_all() works only because the SDK pulls time/net/io-util transitively. fragile - if SDK feature flags change, this breaks. enable explicit features in this crate's Cargo.toml.
foreign/cpp/src/identifier.rs not in this PR's diff but recommend audit for the same magic-number pattern (length: u8 0-sentinel for invalid, etc).
roundtrip tests do not assert polled.partition_id matches the partition the messages were sent to. that's a free regression detector for the u32::MAX-sentinel bug below - add it.

hubcio · 2026-04-29T12:53:52Z

 crate-type = ["staticlib"]

 [dependencies]
+bytes = "1.11.1"


bytes = "1.11.1" hard-pins this crate while the rest of the workspace uses bytes = { workspace = true } (see core/sdk/Cargo.toml:36, core/common/Cargo.toml:40). swap to workspace form so dep versions stay unified.

The cpp sdk is excluded from the workspace, as is the python sdk. That is why this is hard-pinned.

as @slbotbm noted — cpp crate is workspace-excluded (same as python sdk), so the pin stays.

hubcio · 2026-04-29T12:53:52Z

+        let partitioning = match partitioning_kind.as_str() {
+            "balanced" => Partitioning::balanced(),
+            "partition_id" => {
+                if partitioning_value.len() < 4 {


partitioning_value.len() < 4 accepts >4 bytes silently and slices [..4]. caller passing 8 bytes (mistaking u64) gets the low 4 bytes with no error. tighten to != 4 and reject.

tightened to != 4 — 8-byte payloads now error explicitly.

hubcio · 2026-04-29T12:53:52Z

+                })?);
+                Partitioning::partition_id(id)
+            }
+            "messages_key" => Partitioning::messages_key(&partitioning_value).map_err(|error| {


Partitioning::messages_key(&partitioning_value) accepts an empty Vec<u8> at the FFI boundary; the SDK rejects it later. validate at the FFI boundary or at minimum document the constraint - right now the call shape is ambiguous.

rejected at the ffi boundary now.

hubcio · 2026-04-29T12:53:52Z

+        let rust_consumer_id = RustIdentifier::try_from(consumer_id)
+            .map_err(|error| format!("Could not poll messages: {error}"))?;
+
+        let consumer = Consumer {


use Consumer::new(rust_consumer_id) / Consumer::group(rust_consumer_id) instead of the struct literal. matches the SDK idiom (see core/sdk/src/clients/client.rs:138,157) and core/common/src/types/consumer/consumer_kind.rs:78,86 provides both constructors.

switched to Consumer::new() / Consumer::group().

hubcio · 2026-04-29T12:53:52Z

+            }
+        };
+
+        let opt_partition = if partition_id == u32::MAX {


partition_id == u32::MAX is magic number

exposed as ANY_PARTITION_ID with a doc-comment.

hubcio · 2026-04-29T12:53:52Z

+    ASSERT_NO_THROW(iggy::ffi::delete_connection(client));
+}
+
+TEST(LowLevelE2E_Message, SendMessagesBeforeLoginThrows) {


this test (and PollMessagesBeforeLoginThrows at line 479) skip cleanup because they don't create a stream. fine for these specifically. the broader gap: the suite has no fixture for cleanup on early ASSERT failure. when an ASSERT_* trips mid-test, the delete_stream at the end never runs and orphan cpp-msg-* streams accumulate across runs, flaking subsequent suites. add a per-suite TearDown that deletes leftover cpp-msg-* streams. there's already a TODO(slbotbm) for a fixture in test_helpers.hpp:18 - this is what it's for.

I'll do it in future PR's

leaving as-is — @slbotbm said he'd handle it in a follow-up.

hubcio · 2026-04-29T12:53:52Z

+                         "server_default");
+
+    rust::Vec<std::uint8_t> oversized_payload;
+    for (std::uint32_t i = 0; i < 64000001u; i++) {


64MB+1-byte payload via 64M push_back calls is slow - rust::Vec::push_back does no reserve. either pre-reserve with oversized_payload.reserve(64000001) or send exactly MAX_PAYLOAD_SIZE+1 (whatever constant the SDK exposes) instead of a hardcoded 64M.

tried reserve_total but cxx 1.0.194 has it private (caught in CI, reverted in 5d1f1e6). loop now relies on amortised growth — left a comment about it.

hubcio · 2026-04-29T12:53:52Z

+    client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
+                          partition_id_bytes(0), std::move(batch1));
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));


sleep_for(100ms) to ensure the two batches land in distinct timestamps is flaky. IggyTimestamp::now() is microseconds so 100ms should be safe, but on busy CI the boundary can collapse. assert batch2_timestamp - batch1_timestamp >= some_threshold explicitly so the test fails loudly if the gap closes, instead of silently degrading into a tautology.

added an explicit >= 50ms gap assert so it fails loudly instead of silently degrading.

hubcio · 2026-04-29T12:53:52Z

+    auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
+                                        make_numeric_identifier(2), "timestamp", batch2_timestamp, 100, false);
+
+    ASSERT_GE(polled.count, 5u);


ASSERT_GE(polled.count, 5u) then loops polled.messages.size() indexing expected = "batch2-" + i. if the server returns >5 (e.g. boundary inclusion of a batch1 message with equal-or-later timestamp), index 5+ payloads will not be "batch2-N" and the comparison fails with a confusing payload-mismatch error rather than a count error. tighten to ASSERT_EQ(polled.count, 5u) if the contract guarantees it, or assert each polled payload has prefix "batch2-" regardless of count.

switched to prefix matching (batch1- or batch2-) so boundary inclusion doesn't cause confusing payload mismatches.

hubcio · 2026-04-29T12:53:52Z

+    EXPECT_EQ(msg.payload[2], 0x00);
+}
+
+TEST(MessageTest, NewMessageOverwritesPreviousState) {


NewMessageOverwritesPreviousState literally tests "second new_message call zeroes id_lo". this is testing the smell. if new_message is reshaped into a constructor (see lib.rs:191 comment), this test goes away. fine to keep until the API is reshaped, but flagging - it's a test that's locking in awkward behavior.

gone with the api reshape — no more "second call zeros fields" since make_message returns by value.