Merge pull request #171 from influxdata/crepererum/redpanda_22_2_1

fix: update redpanda, fix issues

Author: kodiakhq[bot]

.circleci/config.yml

@@ -148,7 +148,7 @@ jobs:
     # setup multiple docker images (see https://circleci.com/docs/2.0/configuration-reference/#docker)
     docker:
       - image: quay.io/influxdb/rust:ci
-      - image: vectorized/redpanda:v22.1.4
+      - image: vectorized/redpanda:v22.2.1
         name: redpanda-0
         command:
           - redpanda
@@ -161,7 +161,8 @@ jobs:
           - --check=false
           - --kafka-addr redpanda-0:9092
           - --rpc-addr redpanda-0:33145
-      - image: vectorized/redpanda:v22.1.4
+          - --set redpanda.auto_create_topics_enabled=false
+      - image: vectorized/redpanda:v22.2.1
         name: redpanda-1
         command:
           - redpanda
@@ -175,7 +176,8 @@ jobs:
           - --kafka-addr redpanda-1:9092
           - --rpc-addr redpanda-1:33145
           - --seeds redpanda-0:33145
-      - image: vectorized/redpanda:v22.1.4
+          - --set redpanda.auto_create_topics_enabled=false
+      - image: vectorized/redpanda:v22.2.1
         name: redpanda-2
         command:
           - redpanda
@@ -189,6 +191,7 @@ jobs:
           - --kafka-addr redpanda-2:9092
           - --rpc-addr redpanda-2:33145
           - --seeds redpanda-0:33145
+          - --set redpanda.auto_create_topics_enabled=false
       - image: serjs/go-socks5-proxy
         name: proxy
     resource_class: xlarge  # use of a smaller executor tends crashes on link
@@ -243,6 +246,7 @@ jobs:
           - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093
           - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-0:9092,EXTERNAL://kafka-0:9093
           - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
+          - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
       - image: docker.io/bitnami/kafka:3
         name: kafka-1
         environment:
@@ -253,6 +257,7 @@ jobs:
           - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093
           - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-1:9092,EXTERNAL://kafka-1:9093
           - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
+          - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
       - image: docker.io/bitnami/kafka:3
         name: kafka-2
         environment:
@@ -263,6 +268,7 @@ jobs:
           - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093
           - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-2:9092,EXTERNAL://kafka-2:9093
           - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
+          - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
       - image: serjs/go-socks5-proxy
         name: proxy
     resource_class: xlarge  # use of a smaller executor tends crashes on link

README.md

@@ -32,7 +32,7 @@ It will be a good fit for workloads that:
 use rskafka::{
     client::{
         ClientBuilder,
-        partition::Compression,
+        partition::{Compression, UnknownTopicHandling},
     },
     record::Record,
 };
@@ -58,6 +58,7 @@ let partition_client = client
     .partition_client(
         topic.to_owned(),
         0,  // partition
+        UnknownTopicHandling::Retry,
      )
      .await
     .unwrap();

benches/throughput.rs

@@ -22,7 +22,7 @@ use rdkafka::{
 use rskafka::{
     client::{
         consumer::{StartOffset, StreamConsumerBuilder as RsStreamConsumerBuilder},
-        partition::{Compression, PartitionClient},
+        partition::{Compression, PartitionClient, UnknownTopicHandling},
         producer::{aggregator::RecordAggregator, BatchProducerBuilder},
         ClientBuilder,
     },
@@ -448,7 +448,10 @@ async fn setup_rskafka(connection: Vec<String>) -> PartitionClient {
         .await
         .unwrap();
 
-    client.partition_client(topic_name, 0).await.unwrap()
+    client
+        .partition_client(topic_name, 0, UnknownTopicHandling::Retry)
+        .await
+        .unwrap()
 }
 
 static LOG_SETUP: Once = Once::new();

docker-compose-kafka.yml

@@ -22,6 +22,7 @@ services:
       - KAFKA_CFG_LISTENERS=CLIENT://:9000,EXTERNAL://:9010,FOR_PROXY://:9020
       - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-0:9000,EXTERNAL://localhost:9010,FOR_PROXY://kafka-0:9020
       - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
+      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
     volumes:
       - kafka_0_data:/bitnami/kafka
     depends_on:
@@ -38,6 +39,7 @@ services:
       - KAFKA_CFG_LISTENERS=CLIENT://:9000,EXTERNAL://:9011,FOR_PROXY://:9021
       - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-1:9000,EXTERNAL://localhost:9011,FOR_PROXY://kafka-1:9021
       - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
+      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
     volumes:
       - kafka_1_data:/bitnami/kafka
     depends_on:
@@ -54,6 +56,7 @@ services:
       - KAFKA_CFG_LISTENERS=CLIENT://:9000,EXTERNAL://:9012,FOR_PROXY://:9022
       - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-2:9000,EXTERNAL://localhost:9012,FOR_PROXY://kafka-2:9022
       - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
+      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
     volumes:
       - kafka_2_data:/bitnami/kafka
     depends_on:

docker-compose-redpanda.yml

@@ -2,7 +2,7 @@
 version: '3.7'
 services:
   redpanda-0:
-    image: vectorized/redpanda:v22.1.4
+    image: vectorized/redpanda:v22.2.1
     container_name: redpanda-0
     ports:
       - '9010:9010'
@@ -19,8 +19,9 @@ services:
       - --advertise-kafka-addr EXTERNAL://127.0.0.1:9010,FOR_PROXY://redpanda-0:9020
       - --rpc-addr 0.0.0.0:33145
       - --advertise-rpc-addr redpanda-0:33145
+      - --set redpanda.auto_create_topics_enabled=false
   redpanda-1:
-    image: vectorized/redpanda:v22.1.4
+    image: vectorized/redpanda:v22.2.1
     container_name: redpanda-1
     ports:
       - '9011:9011'
@@ -38,8 +39,9 @@ services:
       - --advertise-kafka-addr EXTERNAL://127.0.0.1:9011,FOR_PROXY://redpanda-1:9021
       - --rpc-addr 0.0.0.0:33146
       - --advertise-rpc-addr redpanda-1:33146
+      - --set redpanda.auto_create_topics_enabled=false
   redpanda-2:
-    image: vectorized/redpanda:v22.1.4
+    image: vectorized/redpanda:v22.2.1
     container_name: redpanda-2
     ports:
       - '9012:9012'
@@ -57,6 +59,7 @@ services:
       - --advertise-kafka-addr EXTERNAL://127.0.0.1:9012,FOR_PROXY://redpanda-2:9022
       - --rpc-addr 0.0.0.0:33147
       - --advertise-rpc-addr redpanda-2:33147
+      - --set redpanda.auto_create_topics_enabled=false
   proxy:
     image: serjs/go-socks5-proxy
     ports:

src/client/consumer.rs

@@ -10,14 +10,19 @@
 //!         StartOffset,
 //!         StreamConsumerBuilder,
 //!     },
+//!     partition::UnknownTopicHandling,
 //! };
 //! use std::sync::Arc;
 //!
 //! // get partition client
 //! let connection = "localhost:9093".to_owned();
 //! let client = ClientBuilder::new(vec![connection]).build().await.unwrap();
 //! let partition_client = Arc::new(
-//!     client.partition_client("my_topic", 0).await.unwrap()
+//!     client.partition_client(
+//!         "my_topic",
+//!         0,
+//!         UnknownTopicHandling::Retry,
+//!     ).await.unwrap()
 //! );
 //!
 //! // construct stream consumer

src/client/error.rs

@@ -93,6 +93,9 @@ pub enum Error {
 
     #[error("All retries failed: {0}")]
     RetryFailed(#[from] crate::backoff::BackoffError),
+
+    #[error("Timeout")]
+    Timeout,
 }
 
 impl Error {

src/client/mod.rs

@@ -18,7 +18,7 @@ pub mod producer;
 
 use error::{Error, Result};
 
-use self::controller::ControllerClient;
+use self::{controller::ControllerClient, partition::UnknownTopicHandling};
 
 #[derive(Debug, Error)]
 pub enum ProduceError {
@@ -120,8 +120,15 @@ impl Client {
         &self,
         topic: impl Into<String> + Send,
         partition: i32,
+        unknown_topic_handling: UnknownTopicHandling,
     ) -> Result<PartitionClient> {
-        PartitionClient::new(topic.into(), partition, Arc::clone(&self.brokers)).await
+        PartitionClient::new(
+            topic.into(),
+            partition,
+            Arc::clone(&self.brokers),
+            unknown_topic_handling,
+        )
+        .await
     }
 
     /// Returns a list of topics in the cluster