Repository navigation
[DOCS-39462]-Add a Schema Registry producer example - #5565
Merged
Merged
Conversation
Jamie Gillenwater (jgillenwater)
requested a review
from Robert Yokota (rayokota)
August 10, 2026 17:50
|
🎉 All Contributor License Agreements have been signed. Ready to merge. |
Jamie Gillenwater (jgillenwater)
requested review from
Vahid Fereydouny (fereydunk) and
jmathai-confluent
August 10, 2026 17:52
Jamie Gillenwater (jgillenwater)
marked this pull request as ready for review
August 10, 2026 17:55
Jamie Gillenwater (jgillenwater)
requested a review
from a team
as a code owner
August 10, 2026 17:55
librdkafka has no built-in Schema Registry serializer, so schema-first production in C is done with libserdes plus avro-c. Add an example showing that path, and point to it from the README, so C/C++ readers are not left with raw-bytes producers as the only reference. libserdes and avro-c are not librdkafka dependencies, so the example is deliberately left out of $(EXAMPLES) and built on request with 'make producer_schema_registry', matching how producer_cpp and rdkafka_consume_batch are already handled. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Reformat producer_schema_registry.c with clang-format 18 so it passes the repo style check, and derive the Schema Registry subject from the `topic` variable with snprintf() instead of hardcoding "myTopic-value", so the two cannot drift apart. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
airlock-confluentinc
Bot
force-pushed
the
DOCS-39462
branch
from
September 16, 2026 13:14
ca66aa5 to
d17ed01
Compare
serdes_conf_new() returns NULL if a configuration property is rejected. Passing that NULL to serdes_new() does not crash: serdes_new() treats a NULL conf as "use defaults" and initializes an empty configuration with no Schema Registry URL, so the mistake only surfaces later as an opaque serdes_schema_add() failure. The example also passes (NULL, 0) for errstr, which suppresses the explanation entirely. Rather than add error handling that would clutter the example's main line, call out both points in a comment so readers carrying this into production know to pass a real buffer and check the result. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Robert Yokota (rayokota)
requested changes
Sep 18, 2026
Robert Yokota (rayokota)
left a comment
Member
There was a problem hiding this comment.
libserdes is deprecated. We should have an example using libschemaregistry instead. See https://github.com/confluentinc/libschemaregistry
* Bump bundled OpenSSL to 3.5.8 and libcurl to 8.22.0 for source/autoconf builds * TEMP: drop tag gate on x64/arm64 release-artifact builds to verify dep bump in CI * Revert "TEMP: drop tag gate on x64/arm64 release-artifact builds to verify dep bump in CI" This reverts commit bd573d7. * Update Changelog
…nce starts (#5600) * Tests for KIP-899 and KIP-1102 re-bootstrap cases * Test that a KIP-1102 re-bootstrap actually reaches a bootstrap broker The existing KIP-1102 cases only count "Starting re-bootstrap sequence" log lines. That line is emitted at the top of `rd_kafka_rebootstrap_tmr_cb()`, before any broker is added or connected, so it can't distinguish a working re-bootstrap from an inert one that starts and then keeps querying the very brokers that reported the client's metadata as stale. It also can't be distinguished by address: the harness passes every mock broker in `bootstrap.servers` and all of them are in Metadata, so "asked a bootstrap broker" and "asked a learned broker" are the same host:port on the wire. Add a 6th broker to the `REBOOTSTRAP_REQUIRED` variations and remove it from Metadata responses with `rd_kafka_mock_broker_remove_from_metadata()`. It keeps listening and stays in `bootstrap.servers`, so it is only ever reachable through a bootstrap (`RD_KAFKA_CONFIGURED`) broker object, never as a learned one. A Metadata request arriving there is therefore proof that the sequence resulted in a bootstrap broker being used. Replica assignment already skips brokers that aren't in metadata and clamps the replication factor to the eligible count, so the expected learned broker set is unchanged. Only the error-code variations get the assertion. The ERR__TRANSPORT ones are left at 5 brokers and untouched: there the mock closes the connections, so the client reaches a bootstrap broker through the "all brokers down" path regardless, and the assertion wouldn't be exercising the KIP-1102 path. Request tracking is started at the first action rather than at configuration time, since `rd_kafka_mock_start_request_tracking()` clears the request list and this discards the initial bootstrap requests, which may legitimately have gone to the bootstrap-only broker. The count is sampled from the after-action callback because the harness destroys the mock cluster before returning to the test. Both error-code variations currently fail, reporting 70 re-bootstrap sequences and 0 Metadata requests to the bootstrap-only broker. That is the defect being pinned down, not a test bug: with learned brokers up, `rd_kafka_broker_any_usable()` keeps returning one of them (weight_usable() gives RD_KAFKA_LEARNED a +1000 bonus), so `rd_kafka_connect_any()` is never reached and the re-added bootstrap brokers are never connected. * Fix KIP-1102 re-bootstrap never reaching a bootstrap broker Decommission learned brokers when a re-bootstrap sequence starts so that only the bootstrap servers are used until a Metadata response rebuilds the broker list. Previously the client kept asking the very (still connected) learned brokers that had reported its metadata as stale, since rd_kafka_broker_any_usable() weighs learned brokers higher and always found one usable, so a bootstrap broker was never actually reached. Queued messages on decommissioned brokers are handed back to their partitions and re-sent once new leaders are known from the next Metadata response. Admin requests in flight on a decommissioned broker now fail with RD_KAFKA_RESP_ERR__TRANSPORT (previously RD_KAFKA_RESP_ERR__DESTROY_BROKER) so callers retry instead of treating it as a hard failure. * Fix spurious re-bootstrap and mock cgrp crash from broker decommission rd_kafka_broker_decommission() forced its broker down without marking the failure as planned, so decommissioning several still-up learned brokers at once (as re-bootstrap now does) could itself cross the "all brokers down" threshold and spuriously start a second re-bootstrap sequence. Use rd_kafka_broker_planned_fail() so a decommission is never counted toward ALL_BROKERS_DOWN. Separately, decommissioning the current group coordinator immediately marks it dead and triggers a JoinGroup/SyncGroup retry on a new connection, which can arrive at the mock broker before it has processed the old connection's closure. The mock's per-member response tracking only tolerated one live request per member, asserting on the retry; discard the stale, now-unreachable response instead. * Change the 0151 expectations according to the new policy of removing learned brokers when re-bootstrapping * Bound the 0151 re-bootstrap count by the connect interval, not a measured count do_test_down_then_up_no_rebootstrap_loop asserted 7-12 re-bootstrap sequences, which failed branch-wide on the arm64 local-quick CI jobs with 20 or 21. The count is not a property of the configuration, so no observed range is stable. The subtest's premise was wrong: the attempts are not paced by `reconnect.backoff.ms`. A connect scheduled through rd_kafka_connect_any() ends up in RD_KAFKA_OP_CONNECT, which zeroes rkb_ts_reconnect to expedite the reconnect, so rd_kafka_broker_ reconnect_backoff() returns 0 and the backoff never ramps. The rate instead follows how many subsystems ask for a cluster connection while every broker is down: the sparse-connection keepalive in rd_kafka_1s_tmr_cb(), its deferred one-shot retry, and - the piece that only shows up in CI - ProducerID acquisition, since run_local_quick passes -P and enables idempotence. That third caller takes the rate from ~1/s to ~3/s, giving the 20-21 seen on CI. It reproduces anywhere with 'TESTS=0151 SUBTESTS=down_then_up_no_rebootstrap_loop ./test-runner -P', which yields 9 or 21 roughly half the time each. Assert the one bound that is actually guaranteed: sparse_connect_intvl, max(11, min(reconnect.backoff.ms / 2, 1000)) and so 50ms by default, gates rd_kafka_connect_any(), allowing at most 6000 / 50 = 120 sequences in the 6s window. A loop not gated by connection attempts exceeds that, so the assertion still catches what the subtest is for. Also correct the comment, which described the old pacing theory. * Flaky tests 0118 fix and revert 0151 min count * Test 0151: await a request is sent to the bootstrap-only broker * Avoid adding back the coordinator as learned broker when the response comes from a broker that is being decommissioned * Changelog entry * Upgrade considerations * Report ALL_BROKERS_DOWN at most once per reconnect.backoff.max.ms instead of on every re-bootstrap cycle With metadata.recovery.strategy=rebootstrap, the re-bootstrap timer callback resets every broker's rkb_down_reported flag so that all brokers are retried before the "all brokers down" condition can be declared again. During a sustained outage every retry fails, the threshold is crossed again on each cycle and ALL_BROKERS_DOWN was delivered to the application roughly once per reconnect interval for the whole duration of the outage. Track the time of the last report in rk_last_all_brokers_down_reported_ts and report again only when more than reconnect.backoff.max.ms has passed since it, so a long outage still surfaces periodically but not on every cycle. Reset the timestamp when any broker connection comes up, so a new outage is reported immediately regardless of how recently the previous report was made. The timestamp is compared and updated with a CAS, since the last broker down can be observed concurrently from several broker threads. The re-bootstrap itself is unchanged and still runs on every cycle. Test 0151 do_test_down_then_up_no_rebootstrap_loop now takes the outage duration, polls the handle while the broker is down so errors reach the error callback, and asserts that a 5s and a 15s outage deliver exactly outage / reconnect.backoff.max.ms + 1 errors (1 and 2) over the many re-bootstrap sequences they produce. Fixes #5546 * Don't skip FindCoordinator or Metadata responses from bootstrap brokers that are decommissioning. Fixes failure in 0143_exponential_backoff_mock * Additional changelog entry * test 0151: use atomics to prevent thread sync issues
* Don't commit offsets of a lost assignment A consumer with enable.auto.commit=true could send an OffsetCommit for an assignment it had already lost. On a client-side session timeout, or when max.poll.interval.ms is exceeded, the member id is reset to empty, the assignment is marked lost, and its partitions are revoked. The revoke-time auto-commit of those partitions was still sent because the assignment-lost flag was cleared inside rd_kafka_cgrp_unassign() and rd_kafka_cgrp_incremental_unassign(), before the removed partitions were served, defeating the guard that skips commits for a lost assignment. The commit left with an empty member id and the previous generation. A broker rejects it with UNKNOWN_MEMBER_ID, or, for a static member (group.instance.id set), with FENCED_INSTANCE_ID, which librdkafka raises as a fatal error that stops the consumer. Keep the assignment-lost flag set until the following rejoin applies a new assignment (assign()/incremental_assign() still clear it), so the offsets of a lost assignment are never committed. This restores the behavior documented in rd_kafka_cgrp_incremental_unassign()'s own comment. Add test 0192 (static member, cooperative-sticky, mock) which asserts that no OffsetCommit is sent after a session timeout. Happening since 1.6.0. * Clear assignment lost flag when unassign completes * Add more tests * Add pre-existing missing check for unsubscribe as valid call after assignment lost * Style fix * Update change log
libserdes is deprecated; Confluent's supported C/C++ Schema Registry serde library is libschemaregistry. Replace the C example built on libserdes + avro-c with a C++ example built on libschemaregistry + avro-cpp, modeled on libschemaregistry's own AvroProducer example, and update the Makefile target, .gitignore and docs links accordingly. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This reverts commit b3af96a.
This reverts commit ad276a4.
This reverts commit 2b7f72e.
noramullen1
approved these changes
Sep 21, 2026
Jamie Gillenwater (jgillenwater)
enabled auto-merge (squash)
September 22, 2026 14:30
Jamie Gillenwater (jgillenwater)
requested a review
from Emanuele Sabellico (emasab)
September 29, 2026 14:12
Member
Author
|
Emanuele Sabellico (@emasab) Can you help me troubleshoot this build so we can get these changes merged? |
Emanuele Sabellico (emasab)
approved these changes
Oct 2, 2026
Contributor
|
LGTM! Thanks Jamie Gillenwater (@jgillenwater) ! |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
6 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
examples/producer_schema_registry.c, a producer that registers an Avro schema in Schema Registry and produces a schema-framed message using libserdes and avro-c.producer_schema_registryMakefile target. It is deliberately not in$(EXAMPLES), so plainmakestill works for contributors without libserdes and avro-c installed — same precedent asproducer_cppandrdkafka_consume_batch.examples/README.mdand from the "Usage in code" section of the top-levelREADME.md, noting that librdkafka has no built-in Schema Registry serializer.Addresses DOCS-39462: the sibling clients all now lead with a Schema Registry producer. librdkafka's README carries no inline code example, so instead of swapping a first example this adds the missing Schema Registry example and links to it. This is one of five coordinated PRs across librdkafka, confluent-kafka-python, confluent-kafka-go, confluent-kafka-dotnet, and confluent-kafka-javascript.
Verification
RD_KAFKA_PRODUCER(notRD_KAFKA_TYPE_PRODUCER) confirmed againstsrc/rdkafka.h.