add fail fast producer

allegro · Feb 29, 2024 · dcb847a · dcb847a
1 parent c295537
commit dcb847a
Show file tree

Hide file tree

Showing 5 changed files with 36 additions and 10 deletions.
diff --git a/.../ExperimentalKafkaProducerProperties.java → ...nfig/FailFastKafkaProducerProperties.java b/.../ExperimentalKafkaProducerProperties.java → ...nfig/FailFastKafkaProducerProperties.java
@@ -5,8 +5,8 @@
 
 import java.time.Duration;
 
-@ConfigurationProperties(prefix = "frontend.kafka.experimental-producer")
-public class ExperimentalKafkaProducerProperties implements KafkaProducerParameters {
+@ConfigurationProperties(prefix = "frontend.kafka.fail-fast-producer")
+public class FailFastKafkaProducerProperties implements KafkaProducerParameters {
 
     private Duration speculativeSendDelay = Duration.ofMillis(250);
 
@@ -18,10 +18,12 @@ public class ExperimentalKafkaProducerProperties implements KafkaProducerParamet
 
     private int retries = Integer.MAX_VALUE;
 
-    private Duration retryBackoff = Duration.ofMillis(256);
+    private Duration retryBackoff = Duration.ofMillis(50);
 
     private Duration requestTimeout = Duration.ofMillis(500);
 
+    private Duration deliveryTimeout = Duration.ofMillis(500);
+
     private int batchSize = 16 * 1024;
 
     private int tcpSendBuffer = 128 * 1024;
@@ -160,4 +162,13 @@ public Duration getSpeculativeSendDelay() {
     public void setSpeculativeSendDelay(Duration speculativeSendDelay) {
         this.speculativeSendDelay = speculativeSendDelay;
     }
+
+    @Override
+    public Duration getDeliveryTimeout() {
+        return deliveryTimeout;
+    }
+
+    public void setDeliveryTimeout(Duration deliveryTimeout) {
+        this.deliveryTimeout = deliveryTimeout;
+    }
 }
diff --git a/...d/src/main/java/pl/allegro/tech/hermes/frontend/config/FrontendProducerConfiguration.java b/...d/src/main/java/pl/allegro/tech/hermes/frontend/config/FrontendProducerConfiguration.java
@@ -27,7 +27,7 @@
         SchemaProperties.class,
         KafkaHeaderNameProperties.class,
         KafkaProducerProperties.class,
-        ExperimentalKafkaProducerProperties.class,
+        FailFastKafkaProducerProperties.class,
         KafkaClustersProperties.class,
         HTTPHeadersProperties.class
 })
@@ -55,9 +55,9 @@ public BrokerMessageProducer bufferedMessageBrokerProducer(
 
     @Bean
     public BrokerMessageProducer unbufferedMessageBrokerProducer(
-            @Named("experimentalKafkaMessageProducer") Producers producers,
+            @Named("failFastKafkaProducers") Producers producers,
             MessageToKafkaProducerRecordConverter messageConverter,
-            ExperimentalKafkaProducerProperties kafkaProducerProperties
+            FailFastKafkaProducerProperties kafkaProducerProperties
     ) {
         return new MultiDCKafkaBrokerMessageProducer(producers, new SimpleRemoteProducerProvider(), messageConverter, kafkaProducerProperties.getSpeculativeSendDelay());
     }
@@ -89,10 +89,10 @@ public Producers kafkaMessageProducer(KafkaClustersProperties kafkaClustersPrope
     }
 
     @Bean(destroyMethod = "close")
-    public Producers experimentalKafkaMessageProducer(KafkaClustersProperties kafkaClustersProperties,
-                                          ExperimentalKafkaProducerProperties kafkaProducerProperties,
-                                          LocalMessageStorageProperties localMessageStorageProperties,
-                                          DatacenterNameProvider datacenterNameProvider, BrokerLatencyReporter brokerLatencyReporter) {
+    public Producers failFastKafkaProducers(KafkaClustersProperties kafkaClustersProperties,
+                                            FailFastKafkaProducerProperties kafkaProducerProperties,
+                                            LocalMessageStorageProperties localMessageStorageProperties,
+                                            DatacenterNameProvider datacenterNameProvider, BrokerLatencyReporter brokerLatencyReporter) {
         KafkaProperties kafkaProperties = kafkaClustersProperties.toKafkaProperties(datacenterNameProvider);
         List<KafkaProperties> remoteKafkaProperties = kafkaClustersProperties.toRemoteKafkaProperties(datacenterNameProvider);
         return new KafkaMessageProducerFactory(

diff --git a/...rontend/src/main/java/pl/allegro/tech/hermes/frontend/config/KafkaProducerProperties.java b/...rontend/src/main/java/pl/allegro/tech/hermes/frontend/config/KafkaProducerProperties.java
@@ -20,6 +20,8 @@ public class KafkaProducerProperties implements KafkaProducerParameters {
 
     private Duration requestTimeout = Duration.ofMinutes(30);
 
+    private Duration deliveryTimeout = Duration.ofMinutes(30);
+
     private int batchSize = 16 * 1024;
 
     private int tcpSendBuffer = 128 * 1024;
@@ -150,4 +152,13 @@ public boolean isReportNodeMetricsEnabled() {
     public void setReportNodeMetricsEnabled(boolean reportNodeMetricsEnabled) {
         this.reportNodeMetricsEnabled = reportNodeMetricsEnabled;
     }
+
+    @Override
+    public Duration getDeliveryTimeout() {
+        return deliveryTimeout;
+    }
+
+    public void setDeliveryTimeout(Duration deliveryTimeout) {
+        this.deliveryTimeout = deliveryTimeout;
+    }
 }
diff --git a/...main/java/pl/allegro/tech/hermes/frontend/producer/kafka/KafkaMessageProducerFactory.java b/...main/java/pl/allegro/tech/hermes/frontend/producer/kafka/KafkaMessageProducerFactory.java
@@ -14,6 +14,7 @@
 import static org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.BUFFER_MEMORY_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.COMPRESSION_TYPE_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.LINGER_MS_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.MAX_BLOCK_MS_CONFIG;
@@ -80,6 +81,7 @@ private KafkaProducer<byte[], byte[]> producer(KafkaParameters kafkaParameters,
         props.put(COMPRESSION_TYPE_CONFIG, kafkaProducerParameters.getCompressionCodec());
         props.put(BUFFER_MEMORY_CONFIG, bufferedSizeBytes);
         props.put(REQUEST_TIMEOUT_MS_CONFIG, (int) kafkaProducerParameters.getRequestTimeout().toMillis());
+        props.put(DELIVERY_TIMEOUT_MS_CONFIG, (int) kafkaProducerParameters.getDeliveryTimeout().toMillis());
         props.put(BATCH_SIZE_CONFIG, kafkaProducerParameters.getBatchSize());
         props.put(SEND_BUFFER_CONFIG, kafkaProducerParameters.getTcpSendBuffer());
         props.put(RETRIES_CONFIG, kafkaProducerParameters.getRetries());

diff --git a/...src/main/java/pl/allegro/tech/hermes/frontend/producer/kafka/KafkaProducerParameters.java b/...src/main/java/pl/allegro/tech/hermes/frontend/producer/kafka/KafkaProducerParameters.java
@@ -22,6 +22,8 @@ public interface KafkaProducerParameters {
 
     int getMaxRequestSize();
 
+    Duration getDeliveryTimeout();
+
     Duration getLinger();
 
     Duration getMetricsSampleWindow();