use drainAndShutdown to gracefully shutdown the connection an minize …

…loss of messages

Signed-off-by: Yannic Klem <yannic.klem@bosch.io>

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

@@ -18,6 +18,7 @@
 import java.time.Instant;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 
 import javax.annotation.Nullable;
@@ -293,8 +294,7 @@ private void escalate(final DittoRuntimeException dittoRuntimeException) {
 
         private void stop() {
             if (consumerControl != null) {
-                // TODO use drainAndShutdown?
-                consumerControl.shutdown();
+                consumerControl.drainAndShutdown(new CompletableFuture<>(), materializer.executionContext());
                 consumerControl = null;
             }
         }