karafka-rdkafka syncs + ci sync (#689)

mensfeld · web-flow · commit 154e3d33e1c4 · 2025-09-03T07:29:33.000+02:00
* rules snc

* missing stress

* fixes

* fix rspec exclusions

* remark

* remarks

* remove 3.1

* remarks

* fix

* rm fix

* fix metadata

* Trigger CI build
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,7 @@
 
 ## 0.23.0 (Unreleased)
 - **[Breaking]** Drop support for Ruby 3.1 to move forward with the fiber scheduler work.
+- [Enhancement] Bump librdkafka to `2.11.0`
 - [Enhancement] Support explicit Debian testing due to lib issues.
 - [Enhancement] Support ARM64 Gnu precompilation.
 - [Enhancement] Improve what symbols are exposed outside of the precompiled extensions.
diff --git a/README.md b/README.md
@@ -163,16 +163,17 @@ bundle exec rake produce_messages
 
 | rdkafka-ruby | librdkafka | patches |
 |-|-|-|
-| 0.22.x (2025-07-17) | 2.8.0 (2025-01-07) | yes |
-| 0.21.x (2025-02-13) | 2.8.0 (2025-01-07) | yes |
-| 0.20.0 (2025-01-07) | 2.6.1 (2024-11-18) | yes |
-| 0.19.0 (2024-10-01) | 2.5.3 (2024-09-02) | yes |
-| 0.18.0 (2024-09-02) | 2.5.0 (2024-06-10) | yes |
-| 0.17.0 (2024-08-03) | 2.4.0 (2024-05-07) | no  |
-| 0.16.0 (2024-06-13) | 2.3.0 (2023-10-25) | no  |
-| 0.15.0 (2023-12-03) | 2.3.0 (2023-10-25) | no  |
-| 0.14.0 (2023-11-21) | 2.2.0 (2023-07-12) | no  |
-| 0.13.0 (2023-07-24) | 2.0.2 (2023-01-20) | no  |
-| 0.12.0 (2022-06-17) | 1.9.0 (2022-06-16) | no  |
-| 0.11.0 (2021-11-17) | 1.8.2 (2021-10-18) | no  |
-| 0.10.0 (2021-09-07) | 1.5.0 (2020-07-20) | no  |
+| 0.23.x (Unreleased) | 2.11.0 (2025-07-03) | yes |
+| 0.22.x (2025-07-17) | 2.8.0 (2025-01-07)  | yes |
+| 0.21.x (2025-02-13) | 2.8.0 (2025-01-07)  | yes |
+| 0.20.0 (2025-01-07) | 2.6.1 (2024-11-18)  | yes |
+| 0.19.0 (2024-10-01) | 2.5.3 (2024-09-02)  | yes |
+| 0.18.0 (2024-09-02) | 2.5.0 (2024-06-10)  | yes |
+| 0.17.0 (2024-08-03) | 2.4.0 (2024-05-07)  | no  |
+| 0.16.0 (2024-06-13) | 2.3.0 (2023-10-25)  | no  |
+| 0.15.0 (2023-12-03) | 2.3.0 (2023-10-25)  | no  |
+| 0.14.0 (2023-11-21) | 2.2.0 (2023-07-12)  | no  |
+| 0.13.0 (2023-07-24) | 2.0.2 (2023-01-20)  | no  |
+| 0.12.0 (2022-06-17) | 1.9.0 (2022-06-16)  | no  |
+| 0.11.0 (2021-11-17) | 1.8.2 (2021-10-18)  | no  |
+| 0.10.0 (2021-09-07) | 1.5.0 (2020-07-20)  | no  |
diff --git a/dist/librdkafka-2.11.0.tar.gz b/dist/librdkafka-2.11.0.tar.gz
diff --git a/ext/build_common.sh b/ext/build_common.sh
@@ -19,7 +19,7 @@ readonly CYRUS_SASL_VERSION="2.1.28"
 readonly ZLIB_VERSION="1.3.1"
 readonly ZSTD_VERSION="1.5.7"
 readonly KRB5_VERSION="1.21.3"
-readonly LIBRDKAFKA_VERSION="2.8.0"
+readonly LIBRDKAFKA_VERSION="2.11.0"
 
 # SHA256 checksums for supply chain security
 # Update these when upgrading versions
@@ -29,7 +29,7 @@ declare -A CHECKSUMS=(
     ["zlib-1.3.1.tar.gz"]="9a93b2b7dfdac77ceba5a558a580e74667dd6fede4585b91eefb60f03b72df23"
     ["zstd-${ZSTD_VERSION}.tar.gz"]="eb33e51f49a15e023950cd7825ca74a4a2b43db8354825ac24fc1b7ee09e6fa3"
     ["krb5-${KRB5_VERSION}.tar.gz"]="b7a4cd5ead67fb08b980b21abd150ff7217e85ea320c9ed0c6dadd304840ad35"
-    ["librdkafka-${LIBRDKAFKA_VERSION}.tar.gz"]="5bd1c46f63265f31c6bfcedcde78703f77d28238eadf23821c2b43fc30be3e25"
+    ["librdkafka-${LIBRDKAFKA_VERSION}.tar.gz"]="592a823dc7c09ad4ded1bc8f700da6d4e0c88ffaf267815c6f25e7450b9395ca"
 )
 
 # Colors for output
diff --git a/lib/rdkafka/admin.rb b/lib/rdkafka/admin.rb
@@ -664,7 +664,6 @@ def describe_acl(resource_type:, resource_name:, resource_pattern_type:, princip
       describe_acl_handle
     end
 
-
     # Describe configs
     #
     # @param resources [Array<Hash>]  Array where elements are hashes with two keys:
diff --git a/lib/rdkafka/config.rb b/lib/rdkafka/config.rb
@@ -129,10 +129,7 @@ def self.opaques
     end
 
     # Default config that can be overwritten.
-    DEFAULT_CONFIG = {
-      # Request api version so advanced features work
-      :"api.version.request" => true
-    }.freeze
+    DEFAULT_CONFIG = {}.freeze
 
     # Required config that cannot be overwritten.
     REQUIRED_CONFIG = {
diff --git a/lib/rdkafka/consumer.rb b/lib/rdkafka/consumer.rb
@@ -246,7 +246,7 @@ def assignment_lost?
     # @param timeout_ms [Integer] The timeout for fetching this information.
     # @return [TopicPartitionList]
     # @raise [RdkafkaError] When getting the committed positions fails.
-    def committed(list=nil, timeout_ms=2000)
+    def committed(list=nil, timeout_ms=2_000)
       closed_consumer_check(__method__)
 
       if list.nil?
diff --git a/lib/rdkafka/consumer/topic_partition_list.rb b/lib/rdkafka/consumer/topic_partition_list.rb
@@ -66,10 +66,14 @@ def add_topic(topic, partitions=nil)
       #
       # @param topic [String] The topic's name
       # @param partitions_with_offsets [Hash<Integer, Integer>] The topic's partitions and offsets
+      # @param partitions_with_offsets [Array<Consumer::Partition>] The topic's partitions with offsets
+      #   and metadata (if any)
       #
       # @return [nil]
       def add_topic_and_partitions_with_offsets(topic, partitions_with_offsets)
-        @data[topic.to_s] = partitions_with_offsets.map { |p, o| Partition.new(p, o) }
+        @data[topic.to_s] = partitions_with_offsets.map do |p, o|
+          p.is_a?(Partition) ? p : Partition.new(p, o)
+        end
       end
 
       # Return a `Hash` with the topics as keys and and an array of partition information as the value if present.
diff --git a/lib/rdkafka/producer.rb b/lib/rdkafka/producer.rb
@@ -6,9 +6,6 @@ class Producer
     include Helpers::Time
     include Helpers::OAuth
 
-    # Empty hash used as a default
-    EMPTY_HASH = {}.freeze
-
     # @private
     @@partitions_count_cache = PartitionsCountCache.new
 
@@ -30,6 +27,9 @@ def self.partitions_count_cache=(partitions_count_cache)
       @@partitions_count_cache = partitions_count_cache
     end
 
+    # Empty hash used as a default
+    EMPTY_HASH = {}.freeze
+
     private_constant :EMPTY_HASH
 
     # Raised when there was a critical issue when invoking rd_kafka_topic_new
diff --git a/lib/rdkafka/version.rb b/lib/rdkafka/version.rb
@@ -2,6 +2,6 @@
 
 module Rdkafka
   VERSION = "0.23.0"
-  LIBRDKAFKA_VERSION = "2.8.0"
-  LIBRDKAFKA_SOURCE_SHA256 = "5bd1c46f63265f31c6bfcedcde78703f77d28238eadf23821c2b43fc30be3e25"
+  LIBRDKAFKA_VERSION = "2.11.0"
+  LIBRDKAFKA_SOURCE_SHA256 = "592a823dc7c09ad4ded1bc8f700da6d4e0c88ffaf267815c6f25e7450b9395ca"
 end
diff --git a/spec/lib/rdkafka/admin/delete_topic_handle_spec.rb b/spec/lib/rdkafka/admin/delete_topic_handle_spec.rb
@@ -43,10 +43,12 @@
   describe "#raise_error" do
     let(:pending_handle) { false }
 
+    before { subject[:response] = -1 }
+
     it "should raise the appropriate error" do
       expect {
         subject.raise_error
-      }.to raise_exception(Rdkafka::RdkafkaError, /Success \(no_error\)/)
+      }.to raise_exception(Rdkafka::RdkafkaError, /Unknown broker error \(unknown\)/)
     end
   end
 end
diff --git a/spec/lib/rdkafka/admin_spec.rb b/spec/lib/rdkafka/admin_spec.rb
@@ -34,7 +34,7 @@
   describe '#describe_errors' do
     let(:errors) { admin.class.describe_errors }
 
-    it { expect(errors.size).to eq(170) }
+    it { expect(errors.size).to eq(172) }
     it { expect(errors[-184]).to eq(code: -184, description: 'Local: Queue full', name: '_QUEUE_FULL') }
     it { expect(errors[21]).to eq(code: 21, description: 'Broker: Invalid required acks value', name: 'INVALID_REQUIRED_ACKS') }
   end
@@ -66,7 +66,7 @@
           }.to raise_exception { |ex|
             expect(ex).to be_a(Rdkafka::RdkafkaError)
             expect(ex.message).to match(/Broker: Invalid topic \(topic_exception\)/)
-expect(ex.broker_message).to match(/Topic name.*is invalid: .* contains one or more characters other than ASCII alphanumerics, '.', '_' and '-'/)
+            expect(ex.broker_message).to match(/Topic name.*is invalid: .* contains one or more characters other than ASCII alphanumerics, '.', '_' and '-'/)
           }
         end
       end
@@ -877,7 +877,17 @@
   end
 
   describe '#create_partitions' do
-    let(:metadata) { admin.metadata(topic_name).topics.first }
+    let(:metadata) do
+      begin
+        admin.metadata(topic_name).topics.first
+      rescue Rdkafka::RdkafkaError
+        # We have to wait because if we query too fast after topic creation request, it may not
+        # yet be available throwing an error.
+        # This occurs mostly on slow CIs
+        sleep(1)
+        admin.metadata(topic_name).topics.first
+      end
+    end
 
     context 'when topic does not exist' do
       it 'expect to fail due to unknown partition' do
diff --git a/spec/lib/rdkafka/bindings_spec.rb b/spec/lib/rdkafka/bindings_spec.rb
@@ -170,7 +170,6 @@
   end
 
   describe "oauthbearer callback" do
-
     context "without an oauthbearer callback" do
       it "should do nothing" do
         expect {
diff --git a/spec/lib/rdkafka/config_spec.rb b/spec/lib/rdkafka/config_spec.rb
@@ -159,7 +159,7 @@ def call(config, client_name); end
 
     it "should use default configuration" do
       config = Rdkafka::Config.new
-      expect(config[:"api.version.request"]).to eq true
+      expect(config[:"api.version.request"]).to eq nil
     end
 
     it "should create a consumer with valid config" do
diff --git a/spec/lib/rdkafka/consumer_spec.rb b/spec/lib/rdkafka/consumer_spec.rb
@@ -303,6 +303,7 @@ def send_one_message(val)
         # 2. eat unrelated messages
         while(consumer.poll(timeout)) do; end
       end
+
       after { consumer.unsubscribe }
 
       def send_one_message(val)
@@ -1272,4 +1273,50 @@ def collect(name, list)
       expect(eof_error.code).to eq(:partition_eof)
     end
   end
+
+  describe "long running consumption" do
+    let(:consumer) { rdkafka_consumer_config.consumer }
+    let(:producer) { rdkafka_producer_config.producer }
+
+    after { consumer.close }
+    after { producer.close }
+
+    it "should consume messages continuously for 60 seconds" do
+      consumer.subscribe("consume_test_topic")
+      wait_for_assignment(consumer)
+
+      messages_consumed = 0
+      start_time = Time.now
+
+      # Producer thread - sends message every second
+      producer_thread = Thread.new do
+        counter = 0
+        while Time.now - start_time < 60
+          producer.produce(
+            topic: "consume_test_topic",
+            payload: "payload #{counter}",
+            key: "key #{counter}",
+            partition: 0
+          ).wait
+          counter += 1
+          sleep(1)
+        end
+      end
+
+      # Consumer loop
+      while Time.now - start_time < 60
+        message = consumer.poll(1000)
+        if message
+          expect(message).to be_a Rdkafka::Consumer::Message
+          expect(message.topic).to eq("consume_test_topic")
+          messages_consumed += 1
+          consumer.commit if messages_consumed % 10 == 0
+        end
+      end
+
+      producer_thread.join
+
+      expect(messages_consumed).to be > 50 # Should consume most messages
+    end
+  end
 end
diff --git a/spec/lib/rdkafka/producer/delivery_handle_spec.rb b/spec/lib/rdkafka/producer/delivery_handle_spec.rb
@@ -42,4 +42,19 @@
       end
     end
   end
+
+  describe '#create_result' do
+    let(:pending_handle) { false }
+    let(:report) { subject.create_result }
+
+    context 'when response is 0' do
+      it { expect(report.error).to eq(nil) }
+    end
+
+    context 'when response is not 0' do
+      let(:response) { 1 }
+
+      it { expect(report.error).to eq(Rdkafka::RdkafkaError.new(response)) }
+    end
+  end
 end
diff --git a/spec/lib/rdkafka/producer_spec.rb b/spec/lib/rdkafka/producer_spec.rb
@@ -263,8 +263,6 @@ def call(_, handle)
     expect(message.partition).to eq 1
     expect(message.payload).to eq "payload"
     expect(message.key).to eq "key"
-    # Since api.version.request is on by default we will get
-    # the message creation timestamp if it's not set.
     expect(message.timestamp).to be_within(10).of(Time.now)
   end
 
@@ -401,8 +399,6 @@ def call(_, handle)
     expect(message.partition).to eq 0
     expect(message.payload).to eq "payload"
     expect(message.key).to eq "key"
-    # Since api.version.request is on by default we will get
-    # the message creation timestamp if it's not set.
     expect(message.timestamp).to be_within(10).of(Time.now)
   end
 
@@ -699,16 +695,16 @@ def call(_, handle)
   end
 
   describe '#partition_count' do
-    it { expect(producer.partition_count('consume_test_topic')).to eq(3) }
+    it { expect(producer.partition_count('example_topic')).to eq(1) }
 
     context 'when the partition count value is already cached' do
       before do
-        producer.partition_count('consume_test_topic')
+        producer.partition_count('example_topic')
         allow(::Rdkafka::Metadata).to receive(:new).and_call_original
       end
 
       it 'expect not to query it again' do
-        producer.partition_count('consume_test_topic')
+        producer.partition_count('example_topic')
         expect(::Rdkafka::Metadata).not_to have_received(:new)
       end
     end
@@ -720,25 +716,52 @@ def call(_, handle)
       end
 
       it 'expect to query it again' do
-        producer.partition_count('consume_test_topic')
+        producer.partition_count('example_topic')
         expect(::Rdkafka::Metadata).to have_received(:new)
       end
     end
 
     context 'when the partition count value was cached and time did not expire' do
       before do
         allow(::Process).to receive(:clock_gettime).and_return(0, 29.001)
-        producer.partition_count('consume_test_topic')
+        producer.partition_count('example_topic')
         allow(::Rdkafka::Metadata).to receive(:new).and_call_original
       end
 
       it 'expect not to query it again' do
-        producer.partition_count('consume_test_topic')
+        producer.partition_count('example_topic')
         expect(::Rdkafka::Metadata).not_to have_received(:new)
       end
     end
   end
 
+  describe 'metadata fetch request recovery' do
+    subject(:partition_count) { producer.partition_count('example_topic') }
+
+    describe 'metadata initialization recovery' do
+      context 'when all good' do
+        it { expect(partition_count).to eq(1) }
+      end
+
+      context 'when we fail for the first time with handled error' do
+        before do
+          raised = false
+
+          allow(Rdkafka::Bindings).to receive(:rd_kafka_metadata).and_wrap_original do |m, *args|
+            if raised
+              m.call(*args)
+            else
+              raised = true
+              -185
+            end
+          end
+        end
+
+        it { expect(partition_count).to eq(1) }
+      end
+    end
+  end
+
   describe '#flush' do
     it "should return flush when it can flush all outstanding messages or when no messages" do
       producer.produce(
@@ -761,7 +784,7 @@ def call(_, handle)
 
       after do
         # Allow rdkafka to evict message preventing memory-leak
-        # Set to 5s because slow CIs need time
+        # We give it a bit more time as on slow CIs things take time
         sleep(5)
       end
 
@@ -802,7 +825,7 @@ def call(_, handle)
     context 'when there are outgoing things in the queue' do
       let(:producer) do
         rdkafka_producer_config(
-          "bootstrap.servers": "127.0.0.1:9093",
+          "bootstrap.servers": "127.0.0.1:9095",
           "message.timeout.ms": 2_000
         ).producer
       end

Original file line number	Diff line number	Diff line change
`@@ -664,7 +664,6 @@ def describe_acl(resource_type:, resource_name:, resource_pattern_type:, princip`
`664`	`664`	`describe_acl_handle`
`665`	`665`	`end`
`666`	`666`
`667`		`-`
`668`	`667`	`# Describe configs`
`669`	`668`	`#`
`670`	`669`	`# @param resources [Array<Hash>] Array where elements are hashes with two keys:`