Throttle kafka consumer stream also after consuming a record from Kaf…

…ka (throttling after payload mapping has already been implemented). This early throttling avoids being overloaded with malformed messages and processing them at full speed. Signed-off-by: Dominik Guggemos <dominik.guggemos@bosch.io>
eclipse-ditto · Oct 25, 2021 · 879c323 · 879c323
1 parent f1ccd79
commit 879c323
Show file tree

Hide file tree

Showing 6 changed files with 23 additions and 13 deletions.
diff --git a/...ava/org/eclipse/ditto/connectivity/service/messaging/kafka/AtLeastOnceConsumerStream.java b/...ava/org/eclipse/ditto/connectivity/service/messaging/kafka/AtLeastOnceConsumerStream.java
@@ -21,6 +21,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.eclipse.ditto.base.model.exceptions.DittoRuntimeException;
 import org.eclipse.ditto.connectivity.api.ExternalMessage;
+import org.eclipse.ditto.connectivity.service.config.ConnectionThrottlingConfig;
 import org.eclipse.ditto.connectivity.service.messaging.AcknowledgeableMessage;
 import org.eclipse.ditto.connectivity.service.messaging.monitoring.ConnectionMonitor;
 import org.eclipse.ditto.internal.utils.akka.logging.DittoLoggerFactory;
@@ -55,7 +56,7 @@ final class AtLeastOnceConsumerStream implements KafkaConsumerStream {
     AtLeastOnceConsumerStream(
             final AtLeastOnceKafkaConsumerSourceSupplier sourceSupplier,
             final CommitterSettings committerSettings,
-            final int consumerMaxInflight,
+            final ConnectionThrottlingConfig throttlingConfig,
             final KafkaMessageTransformer kafkaMessageTransformer,
             final boolean dryRun,
             final Materializer materializer,
@@ -87,8 +88,9 @@ final class AtLeastOnceConsumerStream implements KafkaConsumerStream {
         consumerControl = sourceSupplier.get()
                 .filter(committableMessage -> isNotDryRun(committableMessage.record(), dryRun))
                 .map(kafkaMessageTransformer::transform)
+                .throttle(throttlingConfig.getLimit(), throttlingConfig.getInterval())
                 .flatMapConcat(this::processTransformationResult)
-                .mapAsync(consumerMaxInflight, x -> x)
+                .mapAsync(throttlingConfig.getMaxInFlight(), x -> x)
                 .toMat(Committer.sink(committerSettings), Consumer::createDrainingControl)
                 .run(materializer);
     }

diff --git a/...java/org/eclipse/ditto/connectivity/service/messaging/kafka/AtMostOnceConsumerStream.java b/...java/org/eclipse/ditto/connectivity/service/messaging/kafka/AtMostOnceConsumerStream.java
@@ -21,6 +21,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.eclipse.ditto.base.model.exceptions.DittoRuntimeException;
 import org.eclipse.ditto.connectivity.api.ExternalMessage;
+import org.eclipse.ditto.connectivity.service.config.ConnectionThrottlingConfig;
 import org.eclipse.ditto.connectivity.service.messaging.AcknowledgeableMessage;
 import org.eclipse.ditto.connectivity.service.messaging.monitoring.ConnectionMonitor;
 import org.eclipse.ditto.internal.utils.akka.logging.DittoLoggerFactory;
@@ -50,7 +51,7 @@ final class AtMostOnceConsumerStream implements KafkaConsumerStream {
 
     AtMostOnceConsumerStream(
             final AtMostOnceKafkaConsumerSourceSupplier sourceSupplier,
-            final int consumerMaxInflight,
+            final ConnectionThrottlingConfig throttlingConfig,
             final KafkaMessageTransformer kafkaMessageTransformer,
             final boolean dryRun,
             final Materializer materializer,
@@ -81,8 +82,9 @@ final class AtMostOnceConsumerStream implements KafkaConsumerStream {
                 .filter(consumerRecord -> isNotDryRun(consumerRecord, dryRun))
                 .map(kafkaMessageTransformer::transform)
                 .filter(result -> !result.isExpired())
+                .throttle(throttlingConfig.getLimit(), throttlingConfig.getInterval())
                 .flatMapConcat(this::processTransformationResult)
-                .mapAsync(consumerMaxInflight, x -> x)
+                .mapAsync(throttlingConfig.getMaxInFlight(), x -> x)
                 .toMat(Sink.ignore(), Consumer::createDrainingControl)
                 .run(materializer);
     }

diff --git a/...va/org/eclipse/ditto/connectivity/service/messaging/kafka/KafkaConsumerStreamFactory.java b/...va/org/eclipse/ditto/connectivity/service/messaging/kafka/KafkaConsumerStreamFactory.java
@@ -86,7 +86,7 @@ KafkaConsumerStream newAtMostOnceConsumerStream(
 
         final KafkaMessageTransformer kafkaMessageTransformer = buildKafkaMessageTransformer(inboundMonitor);
         return new AtMostOnceConsumerStream(atMostOnceKafkaConsumerSourceSupplier,
-                throttlingConfig.getMaxInFlight(),
+                throttlingConfig,
                 kafkaMessageTransformer,
                 dryRun,
                 materializer,
@@ -105,7 +105,7 @@ KafkaConsumerStream newAtLeastOnceConsumerStream(
         final KafkaMessageTransformer kafkaMessageTransformer = buildKafkaMessageTransformer(inboundMonitor);
         return new AtLeastOnceConsumerStream(atLeastOnceKafkaConsumerSourceSupplier,
                 propertiesFactory.getCommitterSettings(),
-                throttlingConfig.getMaxInFlight(),
+                throttlingConfig,
                 kafkaMessageTransformer,
                 dryRun,
                 materializer,

diff --git a/...service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/TestConstants.java b/...service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/TestConstants.java
@@ -94,7 +94,7 @@
 import org.eclipse.ditto.connectivity.model.signals.commands.query.RetrieveConnectionMetricsResponse;
 import org.eclipse.ditto.connectivity.service.config.ClientConfig;
 import org.eclipse.ditto.connectivity.service.config.ConnectionConfig;
-import org.eclipse.ditto.connectivity.service.config.ConnectionConfigProvider;
+import org.eclipse.ditto.connectivity.service.config.ConnectionThrottlingConfig;
 import org.eclipse.ditto.connectivity.service.config.ConnectivityConfig;
 import org.eclipse.ditto.connectivity.service.config.DittoConnectivityConfig;
 import org.eclipse.ditto.connectivity.service.config.MonitoringConfig;
@@ -161,6 +161,7 @@ public final class TestConstants {
     public static final PingConfig PING_CONFIG;
     public static final ProtocolConfig PROTOCOL_CONFIG;
     public static final MonitoringConfig MONITORING_CONFIG;
+    public static final ConnectionThrottlingConfig KAFKA_THROTTLING_CONFIG;
 
     static {
         final DefaultScopedConfig dittoScopedConfig = DefaultScopedConfig.dittoScoped(CONFIG);
@@ -172,6 +173,7 @@ public final class TestConstants {
         PING_CONFIG = CONNECTIVITY_CONFIG.getPingConfig();
         PROTOCOL_CONFIG = CONNECTIVITY_CONFIG.getProtocolConfig();
         MONITORING_CONFIG = CONNECTIVITY_CONFIG.getMonitoringConfig();
+        KAFKA_THROTTLING_CONFIG = CONNECTION_CONFIG.getKafkaConfig().getConsumerConfig().getThrottlingConfig();
     }
 
     private static final ConnectionType TYPE = ConnectionType.AMQP_10;

diff --git a/...org/eclipse/ditto/connectivity/service/messaging/kafka/AtLeastOnceConsumerStreamTest.java b/...org/eclipse/ditto/connectivity/service/messaging/kafka/AtLeastOnceConsumerStreamTest.java
@@ -29,6 +29,7 @@
 import org.eclipse.ditto.base.model.exceptions.DittoRuntimeException;
 import org.eclipse.ditto.connectivity.api.ExternalMessage;
 import org.eclipse.ditto.connectivity.service.messaging.AcknowledgeableMessage;
+import org.eclipse.ditto.connectivity.service.messaging.TestConstants;
 import org.eclipse.ditto.connectivity.service.messaging.monitoring.ConnectionMonitor;
 import org.junit.After;
 import org.junit.Before;
@@ -91,7 +92,7 @@ public void isImmutable() {
     }
 
     @Test
-    public void appliesBackPressureWhenMessagesAreNotAcknowledged() throws InterruptedException {
+    public void appliesBackPressureWhenMessagesAreNotAcknowledged() {
         new TestKit(actorSystem) {{
             /*
              * Given we have a kafka source which emits records that are all transformed to External messages.
@@ -113,13 +114,14 @@ public void appliesBackPressureWhenMessagesAreNotAcknowledged() throws Interrupt
                     .thenReturn(CommittableTransformationResult.of(result, committableMessage.committableOffset()));
             final ConnectionMonitor connectionMonitor = mock(ConnectionMonitor.class);
             final ConnectionMonitor ackMonitor = mock(ConnectionMonitor.class);
-            final int maxInflight = 10;
+            final int maxInflight = TestConstants.KAFKA_THROTTLING_CONFIG.getMaxInFlight();
             final Materializer materializer = Materializer.createMaterializer(actorSystem);
             final Sink<DittoRuntimeException, TestSubscriber.Probe<DittoRuntimeException>> dreSink =
                     TestSink.create(actorSystem);
 
             // When starting the stream
-            new AtLeastOnceConsumerStream(sourceSupplier, CommitterSettings.apply(actorSystem), maxInflight,
+            new AtLeastOnceConsumerStream(sourceSupplier, CommitterSettings.apply(actorSystem),
+                    TestConstants.KAFKA_THROTTLING_CONFIG,
                     messageTransformer, false, materializer,
                     connectionMonitor, ackMonitor, inboundMappingSink, dreSink);
 

diff --git a/.../org/eclipse/ditto/connectivity/service/messaging/kafka/AtMostOnceConsumerStreamTest.java b/.../org/eclipse/ditto/connectivity/service/messaging/kafka/AtMostOnceConsumerStreamTest.java
@@ -29,6 +29,7 @@
 import org.eclipse.ditto.base.model.exceptions.DittoRuntimeException;
 import org.eclipse.ditto.connectivity.api.ExternalMessage;
 import org.eclipse.ditto.connectivity.service.messaging.AcknowledgeableMessage;
+import org.eclipse.ditto.connectivity.service.messaging.TestConstants;
 import org.eclipse.ditto.connectivity.service.messaging.monitoring.ConnectionMonitor;
 import org.junit.After;
 import org.junit.Before;
@@ -88,7 +89,7 @@ public void isImmutable() {
     }
 
     @Test
-    public void appliesBackPressureWhenMessagesAreNotAcknowledged() throws InterruptedException {
+    public void appliesBackPressureWhenMessagesAreNotAcknowledged() {
         new TestKit(actorSystem) {{
             /*
              * Given we have a kafka source which emits records that are all transformed to External messages.
@@ -104,13 +105,14 @@ public void appliesBackPressureWhenMessagesAreNotAcknowledged() throws Interrupt
             when(messageTransformer.transform(ArgumentMatchers.<ConsumerRecord<String, String>>any())).thenReturn(
                     result);
             final ConnectionMonitor connectionMonitor = mock(ConnectionMonitor.class);
-            final int maxInflight = 10;
+            final int maxInflight = TestConstants.KAFKA_THROTTLING_CONFIG.getMaxInFlight();
             final Materializer materializer = Materializer.createMaterializer(actorSystem);
             final Sink<DittoRuntimeException, TestSubscriber.Probe<DittoRuntimeException>> dreSink =
                     TestSink.create(actorSystem);
 
             // When starting the stream
-            new AtMostOnceConsumerStream(sourceSupplier, maxInflight, messageTransformer, false, materializer,
+            new AtMostOnceConsumerStream(sourceSupplier, TestConstants.KAFKA_THROTTLING_CONFIG, messageTransformer,
+                    false, materializer,
                     connectionMonitor, inboundMappingSink, dreSink);
 
             inboundSinkProbe.ensureSubscription();