allow key and partition key on new topics (#611)

Author: mensfeld

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Rdkafka Changelog
 
 ## 0.21.1 (Unreleased)
+- [Enhancement] Allow for producing to non-existing topics with `key` and `partition_key` present.
 - [Enhancement] Replace TTL-based partition count cache with a global cache that reuses `librdkafka` statistics data when possible.
 - [Enhancement] Support producing and consuming of headers with mulitple values (KIP-82).
 - [Enhancement] Allow native Kafka customization poll time.

lib/rdkafka/producer.rb

@@ -239,6 +239,13 @@ def partition_count(topic)
 
         topic_metadata ? topic_metadata[:partition_count] : -1
       end
+    rescue Rdkafka::RdkafkaError => e
+      # If the topic does not exist, it will be created or if not allowed another error will be
+      # raised. We here return -1 so this can happen without early error happening on metadata
+      # discovery.
+      return -1 if e.code == :unknown_topic_or_part
+
+      raise(e)
     end
 
     # Produces a message to a Kafka topic. The message is added to rdkafka's queue, call {DeliveryHandle#wait wait} on the returned delivery handle to make sure it is delivered.

spec/rdkafka/producer_spec.rb

@@ -364,6 +364,48 @@ def call(_, handle)
     expect(message.key).to eq "key utf8"
   end
 
+  it "should produce a message to a non-existing topic with key and partition key" do
+    new_topic = "it-#{SecureRandom.uuid}"
+
+    handle = producer.produce(
+      # Needs to be a new topic each time
+      topic:   new_topic,
+      payload: "payload",
+      key:     "key",
+      partition_key: "partition_key",
+      label:   "label"
+    )
+
+    # Should be pending at first
+    expect(handle.pending?).to be true
+    expect(handle.label).to eq "label"
+
+    # Check delivery handle and report
+    report = handle.wait(max_wait_timeout: 5)
+    expect(handle.pending?).to be false
+    expect(report).not_to be_nil
+    expect(report.partition).to eq 0
+    expect(report.offset).to be >= 0
+    expect(report.label).to eq "label"
+
+    # Flush and close producer
+    producer.flush
+    producer.close
+
+    # Consume message and verify its content
+    message = wait_for_message(
+      topic: new_topic,
+      delivery_report: report,
+      consumer: consumer
+    )
+    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
+
   context "timestamp" do
     it "should raise a type error if not nil, integer or time" do
       expect {