Switch to KRaft (#385)

docker-compose.yml

@@ -1,23 +1,23 @@
 version: '2'
 
 services:
-  zookeeper:
-    container_name: waterdrop_zookeeper
-    image: confluentinc/cp-zookeeper:7.5.0
-    environment:
-      ZOOKEEPER_CLIENT_PORT: 2181
-
   kafka:
-    container_name: waterdrop_kafka
+    container_name: kafka
     image: confluentinc/cp-kafka:7.5.0
-    depends_on:
-      - zookeeper
+
     ports:
       - 9092:9092
+
     environment:
-      KAFKA_BROKER_ID: 1
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:29092,PLAINTEXT_HOST://localhost:9092
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      CLUSTER_ID: adqr22r231e223131231ww
       KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_PROCESS_ROLES: broker,controller
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://127.0.0.1:9092
+      KAFKA_BROKER_ID: 1
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@127.0.0.1:9093
+      ALLOW_PLAINTEXT_LISTENER: 'yes'
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'

spec/lib/waterdrop/instrumentation/vendors/datadog/metrics_listener_spec.rb

@@ -96,7 +96,7 @@ def initialize
     let(:counts) { dummy_client.buffer[:count] }
     let(:histograms) { dummy_client.buffer[:histogram] }
     let(:guages) { dummy_client.buffer[:gauge] }
-    let(:broker_tag) { { tags: %w[broker:localhost:9092] } }
+    let(:broker_tag) { { tags: %w[broker:127.0.0.1:9092] } }
 
     # We add all expectations in one example not to sleep each time
     it 'expect to have proper metrics in place' do

spec/lib/waterdrop/producer/async_spec.rb

@@ -27,7 +27,11 @@
     end
 
     context 'when producing with good middleware' do
+      let(:message) { build(:valid_message, payload: nil) }
+
       before do
+        producer.produce_sync(topic: message[:topic], payload: nil)
+
         mid = lambda do |msg|
           msg[:partition_key] = '1'
           msg
@@ -36,12 +40,12 @@
         producer.middleware.append mid
       end
 
-      let(:message) { build(:valid_message, payload: nil) }
-
       it { expect(delivery).to be_a(Rdkafka::Producer::DeliveryHandle) }
     end
 
     context 'when producing with corrupted middleware' do
+      let(:message) { build(:valid_message, payload: nil) }
+
       before do
         mid = lambda do |msg|
           msg[:partition_key] = -1
@@ -51,8 +55,6 @@
         producer.middleware.append mid
       end
 
-      let(:message) { build(:valid_message, payload: nil) }
-
       it { expect { delivery }.to raise_error(WaterDrop::Errors::MessageInvalidError) }
     end
 

spec/lib/waterdrop/producer/sync_spec.rb

@@ -58,6 +58,8 @@
 
     let(:message) { build(:valid_message, partition_key: rand.to_s, topic: 'example_topic') }
 
+    before { producer.produce_sync(topic: 'example_topic', payload: '1') }
+
     it { expect(delivery).to be_a(Rdkafka::Producer::DeliveryReport) }
   end