Fix timeout of acknowledgement aggregator

* Before this the ReceiveTimeout did only play a role if only one
  acknowledgement was requested or non of multiple acknowledgements
  were deilvered.
  Because after receiving the first acknowledgement ReceiveTimeout
  will be canceled and therefore the second acknowledgement could take
  much longer

Signed-off-by: Yannic Klem <[email protected]>

Author: Yannic92

internal/models/acks/src/main/java/org/eclipse/ditto/internal/models/acks/AcknowledgementAggregatorActor.java

@@ -48,9 +48,8 @@
 import org.eclipse.ditto.things.model.signals.commands.ThingCommandResponse;
 import org.eclipse.ditto.things.model.signals.commands.ThingErrorResponse;
 
-import akka.actor.AbstractActor;
+import akka.actor.AbstractActorWithTimers;
 import akka.actor.Props;
-import akka.actor.ReceiveTimeout;
 
 /**
  * Actor which is created for an {@code ThingModifyCommand} containing {@code AcknowledgementRequests} responsible for
@@ -59,7 +58,7 @@
  *
  * @since 1.1.0
  */
-public final class AcknowledgementAggregatorActor extends AbstractActor {
+public final class AcknowledgementAggregatorActor extends AbstractActorWithTimers {
 
     private final DittoDiagnosticLoggingAdapter log = DittoLoggerFactory.getDiagnosticLoggingAdapter(this);
 
@@ -84,7 +83,7 @@ private AcknowledgementAggregatorActor(final EntityId entityId,
                 );
 
         timeout = getTimeout(requestCommandHeaders, maxTimeout);
-        getContext().setReceiveTimeout(timeout);
+        timers().startSingleTimer(Control.WAITING_FOR_ACKS_TIMED_OUT, Control.WAITING_FOR_ACKS_TIMED_OUT, timeout);
 
         final Set<AcknowledgementRequest> acknowledgementRequests = requestCommandHeaders.getAcknowledgementRequests();
         ackregator = AcknowledgementAggregator.getInstance(entityId, correlationId, timeout, headerTranslator);
@@ -146,7 +145,7 @@ public Receive createReceive() {
                 .match(Acknowledgement.class, this::handleAcknowledgement)
                 .match(Acknowledgements.class, this::handleAcknowledgements)
                 .match(DittoRuntimeException.class, this::handleDittoRuntimeException)
-                .match(ReceiveTimeout.class, this::handleReceiveTimeout)
+                .match(Control.class, Control.WAITING_FOR_ACKS_TIMED_OUT::equals, this::handleReceiveTimeout)
                 .matchAny(m -> log.warning("Received unexpected message: <{}>", m))
                 .build();
     }
@@ -218,7 +217,7 @@ private static Optional<JsonValue> getPayload(final CommandResponse<?> response)
         return result;
     }
 
-    private void handleReceiveTimeout(final ReceiveTimeout receiveTimeout) {
+    private void handleReceiveTimeout(final Control receiveTimeout) {
         log.withCorrelationId(correlationId).info("Timed out waiting for all requested acknowledgements, " +
                 "completing Acknowledgements with timeouts...");
         completeAcknowledgements(null);
@@ -321,4 +320,8 @@ private static Duration getTimeout(final DittoHeaders headers, final Duration ma
                 .orElse(maxTimeout);
     }
 
+    private enum Control {
+        WAITING_FOR_ACKS_TIMED_OUT;
+    }
+
 }

internal/models/acks/src/test/java/org/eclipse/ditto/internal/models/acks/AcknowledgementAggregatorActorTest.java

@@ -15,6 +15,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.time.Duration;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
 import org.eclipse.ditto.base.model.acks.AcknowledgementLabel;
@@ -23,12 +24,12 @@
 import org.eclipse.ditto.base.model.common.ResponseType;
 import org.eclipse.ditto.base.model.headers.DittoHeaderDefinition;
 import org.eclipse.ditto.base.model.headers.DittoHeaders;
-import org.eclipse.ditto.internal.models.acks.config.DefaultAcknowledgementConfig;
-import org.eclipse.ditto.things.model.ThingId;
-import org.eclipse.ditto.protocol.HeaderTranslator;
 import org.eclipse.ditto.base.model.signals.acks.Acknowledgement;
 import org.eclipse.ditto.base.model.signals.acks.Acknowledgements;
 import org.eclipse.ditto.base.model.signals.commands.exceptions.GatewayCommandTimeoutException;
+import org.eclipse.ditto.internal.models.acks.config.DefaultAcknowledgementConfig;
+import org.eclipse.ditto.protocol.HeaderTranslator;
+import org.eclipse.ditto.things.model.ThingId;
 import org.eclipse.ditto.things.model.signals.commands.ThingErrorResponse;
 import org.eclipse.ditto.things.model.signals.commands.modify.DeleteThing;
 import org.eclipse.ditto.things.model.signals.commands.modify.DeleteThingResponse;
@@ -188,6 +189,49 @@ public void keepCommandHeaders() {
                             .build()
             );
             assertThat(acks.getSize()).isEqualTo(2);
+            assertThat(acks.getAcknowledgement(label1).map(Acknowledgement::getHttpStatus)).contains(
+                    HttpStatus.UNAUTHORIZED);
+            assertThat(acks.getAcknowledgement(label2).map(Acknowledgement::getHttpStatus)).contains(
+                    HttpStatus.PAYMENT_REQUIRED);
+        }};
+    }
+
+    @Test
+    public void awaitsOnlyUntilTimeout() throws InterruptedException {
+        new TestKit(actorSystem) {{
+            // GIVEN
+            final Duration timeout = Duration.ofSeconds(5);
+            final String tag = "tag";
+            final String correlationId = "awaitsOnlyUntilTimeout";
+            final ThingId thingId = ThingId.of("thing:id");
+            final AcknowledgementLabel label1 = AcknowledgementLabel.of("ack1");
+            final AcknowledgementLabel label2 = AcknowledgementLabel.of("ack2");
+            final ThingModifyCommand<?> command = DeleteThing.of(thingId, DittoHeaders.newBuilder()
+                    .correlationId(correlationId)
+                    .timeout(timeout)
+                    .acknowledgementRequest(AcknowledgementRequest.of(label1), AcknowledgementRequest.of(label2))
+                    .build());
+            final ActorRef underTest = childActorOf(getAcknowledgementAggregatorProps(command, this));
+
+            // WHEN
+            final Acknowledgement ack1 = Acknowledgement.of(label1, thingId, HttpStatus.UNAUTHORIZED,
+                    DittoHeaders.newBuilder().correlationId(correlationId).putHeader(tag, label1.toString()).build());
+            final Acknowledgement ack2 = Acknowledgement.of(label2, thingId, HttpStatus.PAYMENT_REQUIRED,
+                    DittoHeaders.newBuilder().correlationId(correlationId).putHeader(tag, label2.toString()).build());
+            TimeUnit.SECONDS.sleep(
+                    timeout.toSeconds() / 2 + 1); // Wait more than half the time before sending first ack
+            underTest.tell(ack1, ActorRef.noSender());
+            TimeUnit.SECONDS.sleep(timeout.toSeconds() / 2 +
+                    1); // Wait more than half the time before sending second ack. This should not be taken into account.
+            underTest.tell(ack2, ActorRef.noSender());
+
+            // THEN
+            final Acknowledgements acks = expectMsgClass(Acknowledgements.class);
+            assertThat(acks.getSize()).isEqualTo(2);
+            assertThat(acks.getAcknowledgement(label1).map(Acknowledgement::getHttpStatus)).contains(
+                    HttpStatus.UNAUTHORIZED);
+            assertThat(acks.getAcknowledgement(label2).map(Acknowledgement::getHttpStatus)).contains(
+                    HttpStatus.REQUEST_TIMEOUT);
         }};
     }