remove throttling limits currently in place by default for Ditto:

* Kafka consumer was limited to receive 100 msg/s
* AMQP 1.0 consumer was limited to receive 100 msg/s
* Websocket was limited to receive 100 msg/s

Signed-off-by: Thomas Jaeckle <thomas.jaeckle@bosch.io>

base/service/src/main/java/org/eclipse/ditto/base/service/config/ThrottlingConfig.java

@@ -75,7 +75,7 @@ static ThrottlingConfig of(final Config config) {
 
     /**
      * An enumeration of the known config path expressions and their associated default values for
-     * {@code Amqp10Config}.
+     * {@code ThrottlingConfig}.
      */
     enum ConfigValue implements KnownConfigValue {
 
@@ -89,7 +89,7 @@ enum ConfigValue implements KnownConfigValue {
          * The consumer throttling limit defining processed messages per configured
          * {@link #INTERVAL interval}.
          */
-        LIMIT("limit", 100);
+        LIMIT("limit", 0);
 
         private final String path;
         private final Object defaultValue;

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/config/Amqp10ConsumerConfig.java

@@ -67,7 +67,7 @@ enum ConfigValue implements KnownConfigValue {
         /**
          * Whether consumer rate limit is enabled.
          */
-        RATE_LIMIT_ENABLED("rate-limit-enabled", true),
+        RATE_LIMIT_ENABLED("rate-limit-enabled", false),
 
         /**
          * When to forget messages for which redelivery was requested (they may be consumed by another consumer).

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/InboundMappingSink.java

@@ -141,7 +141,8 @@ private Sink<Object, NotUsed> mapMessage() {
                         );
 
         final Flow<Object, InboundMappingOutcomes, NotUsed> flowWithOptionalThrottling;
-        if (throttlingConfig != null) {
+        if (throttlingConfig != null &&
+                throttlingConfig.getLimit() > 0 && throttlingConfig.getInterval().negated().isNegative()) {
             flowWithOptionalThrottling = mapMessageFlow
                     .throttle(throttlingConfig.getLimit(), throttlingConfig.getInterval(),
                             outcomes -> (int) outcomes.getOutcomes()

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/kafka/AtLeastOnceConsumerStream.java

@@ -13,6 +13,7 @@
 package org.eclipse.ditto.connectivity.service.messaging.kafka;
 
 import java.nio.ByteBuffer;
+import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.BiConsumer;
@@ -90,10 +91,21 @@ final class AtLeastOnceConsumerStream implements KafkaConsumerStream {
                 .run(materializer);
 
         this.materializer = materializer;
-        consumerControl = sourceSupplier.get()
+
+        final var source = sourceSupplier.get()
                 .filter(committableMessage -> isNotDryRun(committableMessage.record(), dryRun))
-                .map(kafkaMessageTransformer::transform)
-                .throttle(throttlingConfig.getLimit(), throttlingConfig.getInterval())
+                .map(kafkaMessageTransformer::transform);
+
+        final int throttlingLimit = throttlingConfig.getLimit();
+        final Duration throttlingInterval = throttlingConfig.getInterval();
+        final Source<CommittableTransformationResult, Consumer.Control> throttledSource;
+        if (throttlingLimit > 0 && throttlingInterval.negated().isNegative()) {
+            throttledSource = source.throttle(throttlingLimit, throttlingInterval);
+        } else {
+            throttledSource = source;
+        }
+
+        consumerControl = throttledSource
                 .flatMapConcat(this::processTransformationResult)
                 .mapAsync(throttlingConfig.getMaxInFlight(), x -> x)
                 .toMat(Committer.sink(committerSettings), Consumer::createDrainingControl)

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/kafka/AtMostOnceConsumerStream.java

@@ -13,6 +13,7 @@
 package org.eclipse.ditto.connectivity.service.messaging.kafka;
 
 import java.nio.ByteBuffer;
+import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.BiConsumer;
@@ -83,11 +84,22 @@ final class AtMostOnceConsumerStream implements KafkaConsumerStream {
                                 "Please contact the service team", result
                 )))
                 .run(materializer);
-        consumerControl = sourceSupplier.get()
+
+        final var source = sourceSupplier.get()
                 .filter(consumerRecord -> isNotDryRun(consumerRecord, dryRun))
                 .map(kafkaMessageTransformer::transform)
-                .filter(result -> !result.isExpired())
-                .throttle(throttlingConfig.getLimit(), throttlingConfig.getInterval())
+                .filter(result -> !result.isExpired());
+
+        final int throttlingLimit = throttlingConfig.getLimit();
+        final Duration throttlingInterval = throttlingConfig.getInterval();
+        final Source<TransformationResult, Consumer.Control> throttledSource;
+        if (throttlingLimit > 0 && throttlingInterval.negated().isNegative()) {
+            throttledSource = source.throttle(throttlingLimit, throttlingInterval);
+        } else {
+            throttledSource = source;
+        }
+
+        consumerControl = throttledSource
                 .flatMapConcat(this::processTransformationResult)
                 .mapAsync(throttlingConfig.getMaxInFlight(), x -> x)
                 .toMat(Sink.ignore(), Consumer::createDrainingControl)

connectivity/service/src/main/resources/connectivity.conf

@@ -165,7 +165,7 @@ ditto {
       amqp10 {
         consumer {
           # Whether rate limit according to throughput and acknowledgement is enabled.
-          rate-limit-enabled = true
+          rate-limit-enabled = false
           rate-limit-enabled = ${?AMQP10_CONSUMER_RATE_LIMIT_ENABLED}
 
           # When to forget an NAcked redelivery=true message -- those may be consumed by another consumer
@@ -361,13 +361,15 @@ ditto {
         consumer {
           throttling {
             # Interval at which the consumer is throttled. Values smaller than 1s are treated as 1s.
+            # Disable throttling with a value of zero.
             interval = 1s
             interval = ${?KAFKA_CONSUMER_THROTTLING_INTERVAL}
 
             # The maximum number of messages the consumer is allowed to receive within the configured
             # throttling interval e.g. 100 msgs/s.
-            # Values smaller than 1 are treated as 1.
-            limit = 100
+            # Values smaller than 0 are treated as 0.
+            # Disable throttling with a value of zero.
+            limit = 0
             limit = ${?KAFKA_CONSUMER_THROTTLING_LIMIT}
 
             # Throttling detection tolerance in percent e.g. for a tolerance of 5% if consumed messages are above

gateway/service/src/main/resources/gateway.conf

@@ -121,7 +121,7 @@ ditto {
 
           # The maximum number of messages the websocket is allowed to receive within the configured
           # throttling interval e.g. 100 msgs/s. Disable throttling with a value of zero.
-          limit = 100
+          limit = 0
           limit = ${?GATEWAY_WEBSOCKET_THROTTLING_LIMIT}
         }
       }
@@ -134,7 +134,7 @@ ditto {
 
           # The maximum number of things a search SSE connection is allowed to send within the configured
           # throttling interval e.g. 100 things/s. Disable throttling with a value of zero.
-          limit = 100
+          limit = 0
           limit = ${?GATEWAY_SSE_THROTTLING_LIMIT}
         }
       }