Count successful and failed acknowledgements in connection metrics

Signed-off-by: Yannic Klem <yannic.klem@bosch.io>
eclipse-ditto · Sep 3, 2021 · ece8ad3 · ece8ad3
1 parent d6a7fb8
commit ece8ad3
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 6 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
@@ -46,6 +46,7 @@ final class AtLeastOnceConsumerStream implements KafkaConsumerStream {
     private static final Logger LOGGER = DittoLoggerFactory.getThreadSafeLogger(AtLeastOnceConsumerStream.class);
 
     private final ConnectionMonitor inboundMonitor;
+    private final ConnectionMonitor ackMonitor;
     private final Materializer materializer;
     private final Sink<AcknowledgeableMessage, NotUsed> inboundMappingSink;
     private final Sink<DittoRuntimeException, ?> dreSink;
@@ -59,10 +60,12 @@ final class AtLeastOnceConsumerStream implements KafkaConsumerStream {
             final boolean dryRun,
             final Materializer materializer,
             final ConnectionMonitor inboundMonitor,
+            final ConnectionMonitor ackMonitor,
             final Sink<AcknowledgeableMessage, NotUsed> inboundMappingSink,
             final Sink<DittoRuntimeException, ?> dreSink) {
 
         this.inboundMonitor = inboundMonitor;
+        this.ackMonitor = ackMonitor;
         this.inboundMappingSink = inboundMappingSink;
         this.dreSink = dreSink;
         this.materializer = materializer;
@@ -91,7 +94,7 @@ private Source<CompletableFuture<CommittableOffset>, NotUsed> processTransformat
             final CommittableTransformationResult result) {
         if (isExternalMessage(result)) {
             return Source.single(result)
-                    .map(AtLeastOnceConsumerStream::toAcknowledgeableMessage)
+                    .map(this::toAcknowledgeableMessage)
                     .alsoTo(this.externalMessageSink())
                     .map(KafkaAcknowledgableMessage::getAcknowledgementFuture);
         }
@@ -122,12 +125,12 @@ private static boolean isExternalMessage(final CommittableTransformationResult t
         return transformationResult.getTransformationResult().getExternalMessage().isPresent();
     }
 
-    private static KafkaAcknowledgableMessage toAcknowledgeableMessage(final CommittableTransformationResult value) {
+    private KafkaAcknowledgableMessage toAcknowledgeableMessage(final CommittableTransformationResult value) {
         final ExternalMessage externalMessage = value.getTransformationResult()
                 .getExternalMessage()
                 .orElseThrow(); // at this point, the ExternalMessage is present
         final CommittableOffset committableOffset = value.getCommittableOffset();
-        return new KafkaAcknowledgableMessage(externalMessage, committableOffset);
+        return new KafkaAcknowledgableMessage(externalMessage, committableOffset, ackMonitor);
     }
 
     private static boolean isNotDryRun(final ConsumerRecord<String, String> cRecord, final boolean dryRun) {

diff --git a/...va/org/eclipse/ditto/connectivity/service/messaging/kafka/KafkaAcknowledgableMessage.java b/...va/org/eclipse/ditto/connectivity/service/messaging/kafka/KafkaAcknowledgableMessage.java
@@ -19,6 +19,7 @@
 
 import org.eclipse.ditto.connectivity.api.ExternalMessage;
 import org.eclipse.ditto.connectivity.service.messaging.AcknowledgeableMessage;
+import org.eclipse.ditto.connectivity.service.messaging.monitoring.ConnectionMonitor;
 
 import akka.kafka.ConsumerMessage;
 
@@ -38,14 +39,20 @@ final class KafkaAcknowledgableMessage {
     private final CompletableFuture<ConsumerMessage.CommittableOffset> acknowledgementFuture;
 
     KafkaAcknowledgableMessage(final ExternalMessage message,
-            final ConsumerMessage.CommittableOffset committableOffset) {
+            final ConsumerMessage.CommittableOffset committableOffset,
+            final ConnectionMonitor ackMonitor) {
         this.acknowledgementFuture = new CompletableFuture<>();
         this.acknowledgeableMessage = AcknowledgeableMessage.of(message,
-                () -> acknowledgementFuture.complete(committableOffset),
+                () -> {
+                    ackMonitor.success(message);
+                    acknowledgementFuture.complete(committableOffset);
+                },
                 shouldRedeliver -> {
                     if (shouldRedeliver) {
+                        ackMonitor.exception("Message was rejected and redelivery is requested.");
                         acknowledgementFuture.completeExceptionally(MessageRejectedException.getInstance());
                     } else {
+                        ackMonitor.exception("Message was rejected and no redelivery is requested.");
                         acknowledgementFuture.complete(committableOffset);
                     }
                 });

diff --git a/.../main/java/org/eclipse/ditto/connectivity/service/messaging/kafka/KafkaConsumerActor.java b/.../main/java/org/eclipse/ditto/connectivity/service/messaging/kafka/KafkaConsumerActor.java
@@ -71,7 +71,8 @@ private KafkaConsumerActor(final Connection connection,
                     () -> {
                         final KafkaConsumerStream kafkaConsumerStream =
                                 streamFactory.newAtLeastOnceConsumerStream(materializer, inboundMonitor,
-                                        getMessageMappingSink(), getDittoRuntimeExceptionSink());
+                                        inboundAcknowledgedMonitor, getMessageMappingSink(),
+                                        getDittoRuntimeExceptionSink());
                         kafkaConsumerStream.whenComplete(this::handleStreamCompletion);
                         return kafkaConsumerStream;
                     }, exponentialBackOffConfig);

diff --git a/...va/org/eclipse/ditto/connectivity/service/messaging/kafka/KafkaConsumerStreamFactory.java b/...va/org/eclipse/ditto/connectivity/service/messaging/kafka/KafkaConsumerStreamFactory.java
@@ -97,6 +97,7 @@ KafkaConsumerStream newAtMostOnceConsumerStream(
     KafkaConsumerStream newAtLeastOnceConsumerStream(
             final Materializer materializer,
             final ConnectionMonitor inboundMonitor,
+            final ConnectionMonitor ackMonitor,
             final Sink<AcknowledgeableMessage, NotUsed> messageMappingSink,
             final Sink<DittoRuntimeException, ?> dreSink) {
 
@@ -108,6 +109,7 @@ KafkaConsumerStream newAtLeastOnceConsumerStream(
                 dryRun,
                 materializer,
                 inboundMonitor,
+                ackMonitor,
                 messageMappingSink,
                 dreSink);
     }