add JmsOperationTimedOutException to user-indicated-errors-base;

add catch block for JmsOperationTimedOutException in AmqpClientActor and log a warning in this special case; fix sonar issues; Signed-off-by: Stefan Maute <stefan.maute@bosch.io>
eclipse-ditto · May 5, 2022 · 5396893 · 5396893
1 parent d7a1672
commit 5396893
Show file tree

Hide file tree

Showing 94 changed files with 239 additions and 285 deletions.
diff --git a/...n/java/org/eclipse/ditto/connectivity/service/config/ConnectionConfigProviderFactory.java b/...n/java/org/eclipse/ditto/connectivity/service/config/ConnectionConfigProviderFactory.java
@@ -15,7 +15,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.function.Predicate;
-import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
 import org.atteo.classindex.ClassIndex;
@@ -106,7 +105,7 @@ private static Class<? extends ConnectionConfigProvider> findProviderClass(
         final List<Class<? extends ConnectionConfigProvider>> candidates =
                 StreamSupport.stream(subclasses.spliterator(), false)
                         .filter(classPredicate)
-                        .collect(Collectors.toList());
+                        .toList();
 
         if (candidates.size() == 1) {
             return candidates.get(0);

diff --git a/...main/java/org/eclipse/ditto/connectivity/service/mapping/DefaultMessageMapperFactory.java b/...main/java/org/eclipse/ditto/connectivity/service/mapping/DefaultMessageMapperFactory.java
@@ -239,12 +239,12 @@ private static List<MessageMapperExtension> loadMessageMapperExtensions(final Dy
             final ClassTag<MessageMapperExtension> tag =
                     scala.reflect.ClassTag$.MODULE$.apply(MessageMapperExtension.class);
             return dynamicAccess.createInstanceFor(clazz, List$.MODULE$.empty(), tag).get();
-        }).collect(Collectors.toList());
+        }).toList();
     }
 
     private static List<Class<? extends MessageMapperExtension>> loadMessageMapperExtensionClasses() {
         return StreamSupport.stream(ClassIndex.getSubclasses(MessageMapperExtension.class).spliterator(), false)
-                .collect(Collectors.toList());
+                .toList();
     }
 
     /**

diff --git a/...ain/java/org/eclipse/ditto/connectivity/service/mapping/DefaultMessageMapperRegistry.java b/...ain/java/org/eclipse/ditto/connectivity/service/mapping/DefaultMessageMapperRegistry.java
@@ -18,7 +18,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.stream.Collectors;
 
 import javax.annotation.Nullable;
 
@@ -67,7 +66,7 @@ public List<MessageMapper> getMappers(final PayloadMapping payloadMapping) {
         return payloadMapping.getMappings().stream()
                 .map(this::resolveMessageMapper)
                 .map(resolvedMapper -> null == resolvedMapper ? defaultMapper : resolvedMapper)
-                .collect(Collectors.toList());
+                .toList();
     }
 
     @Nullable

diff --git a/...e/src/main/java/org/eclipse/ditto/connectivity/service/mapping/WrappingMessageMapper.java b/...e/src/main/java/org/eclipse/ditto/connectivity/service/mapping/WrappingMessageMapper.java
@@ -19,7 +19,6 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.function.UnaryOperator;
-import java.util.stream.Collectors;
 
 import org.eclipse.ditto.base.model.headers.DittoHeaders;
 import org.eclipse.ditto.connectivity.api.ExternalMessage;
@@ -119,7 +118,7 @@ public List<ExternalMessage> map(final Adaptable adaptable) {
 
         return externalMessages.stream()
                 .map(markAsResponse)
-                .collect(Collectors.toList());
+                .toList();
     }
 
     private <T> List<T> checkMaxMappedMessagesLimit(final List<T> mappingResult, final int maxMappedMessages,

diff --git a/...rvice/src/main/java/org/eclipse/ditto/connectivity/service/messaging/BaseClientActor.java b/...rvice/src/main/java/org/eclipse/ditto/connectivity/service/messaging/BaseClientActor.java
@@ -1384,7 +1384,7 @@ private FSM.State<BaseClientState, BaseClientData> retrieveConnectionStatus(fina
         final List<ActorRef> childrenToAsk = StreamSupport.stream(getContext().getChildren().spliterator(), false)
                 .filter(child -> noAddressReportingChildNamePatterns.stream()
                         .noneMatch(p -> p.matcher(child.path().name()).matches()))
-                .collect(Collectors.toList());
+                .toList();
         final ConnectivityStatus clientConnectionStatus = data.getConnectionStatus();
         if (childrenToAsk.size() != expectedNumberOfChildren) {
             if (clientConnectionStatus.isFailure() || clientConnectionStatus == ConnectivityStatus.UNKNOWN) {
@@ -1730,7 +1730,7 @@ private Pair<ActorRef, ActorRef> startOutboundActors(final ProtocolAdapter proto
                     proxyActorSelection, protocolAdapter, logger);
             outboundMappingProcessors = IntStream.range(0, processorPoolSize)
                     .mapToObj(i -> OutboundMappingProcessor.of(settings))
-                    .collect(Collectors.toList());
+                    .toList();
         } catch (final DittoRuntimeException dre) {
             connectionLogger.failure("Failed to start message mapping processor due to: {0}", dre.getMessage());
             logger.info("Got DittoRuntimeException during initialization of MessageMappingProcessor: {} {} - desc: {}",

diff --git a/...ice/src/main/java/org/eclipse/ditto/connectivity/service/messaging/BaseConsumerActor.java b/...ice/src/main/java/org/eclipse/ditto/connectivity/service/messaging/BaseConsumerActor.java
@@ -150,11 +150,11 @@ private void prepareResponseHandler(final AcknowledgeableMessage acknowledgeable
         final Duration askTimeout = acknowledgementConfig.getCollectorFallbackAskTimeout();
         // Ask response collector actor to get the collected responses in a future
         Patterns.ask(responseCollector, ResponseCollectorActor.query(), askTimeout).thenCompose(output -> {
-            if (output instanceof ResponseCollectorActor.Output) {
-                return CompletableFuture.completedFuture((ResponseCollectorActor.Output) output);
-            } else if (output instanceof Throwable) {
+            if (output instanceof ResponseCollectorActor.Output responseCollectorActorOutput) {
+                return CompletableFuture.completedFuture(responseCollectorActorOutput);
+            } else if (output instanceof Throwable throwable) {
                 log().debug("Patterns.ask failed. ResponseCollector=<{}>", responseCollector);
-                return CompletableFuture.failedFuture((Throwable) output);
+                return CompletableFuture.failedFuture(throwable);
             } else {
                 log().error("Expect ResponseCollectorActor.Output, got: <{}>. ResponseCollector=<{}>", output,
                         responseCollector);
@@ -288,8 +288,8 @@ private static boolean someFailedResponseRequiresRedelivery(final Collection<Com
     }
 
     private static Stream<? extends CommandResponse<?>> extractAggregatedResponses(final CommandResponse<?> response) {
-        if (response instanceof Acknowledgements) {
-            return ((Acknowledgements) response).stream();
+        if (response instanceof Acknowledgements acknowledgements) {
+            return acknowledgements.stream();
         } else {
             return Stream.of(response);
         }

diff --git a/...ce/src/main/java/org/eclipse/ditto/connectivity/service/messaging/BasePublisherActor.java b/...ce/src/main/java/org/eclipse/ditto/connectivity/service/messaging/BasePublisherActor.java
@@ -139,7 +139,7 @@ protected BasePublisherActor(final Connection connection,
         responseDroppedMonitor = connectionMonitorRegistry.forResponseDropped(connection);
         responsePublishedMonitor = connectionMonitorRegistry.forResponsePublished(connection);
         responseAcknowledgedMonitor = connectionMonitorRegistry.forResponseAcknowledged(connection);
-        replyTargets = connection.getSources().stream().map(Source::getReplyTarget).collect(Collectors.toList());
+        replyTargets = connection.getSources().stream().map(Source::getReplyTarget).toList();
         acknowledgementSizeBudget = connectionConfig.getAcknowledgementConfig().getIssuedMaxBytes();
         logger = DittoLoggerFactory.getThreadSafeDittoLoggingAdapter(this)
                 .withMdcEntry(ConnectivityMdcEntryKey.CONNECTION_ID, connection.getId());
@@ -189,8 +189,8 @@ private void sendMultiMappedOutboundSignal(final OutboundSignal.MultiMapped mult
                     final Collection<Acknowledgement> acknowledgements = new ArrayList<>();
                     final Collection<CommandResponse<?>> nonAcknowledgements = new ArrayList<>();
                     responsesList.forEach(response -> {
-                        if (response instanceof Acknowledgement) {
-                            acknowledgements.add((Acknowledgement) response);
+                        if (response instanceof Acknowledgement acknowledgement) {
+                            acknowledgements.add(acknowledgement);
                         } else {
                             nonAcknowledgements.add(response);
                         }
@@ -258,7 +258,7 @@ private static <T> CompletionStage<List<T>> aggregateNonNullFutures(final Comple
                 .thenApply(aVoid -> Arrays.stream(futures)
                         .map(CompletableFuture::join)
                         .filter(Objects::nonNull)
-                        .collect(Collectors.toList()));
+                        .toList());
     }
 
     private Acknowledgements appendConnectionId(final Acknowledgements acknowledgements) {
@@ -269,7 +269,7 @@ private static Acknowledgements appendConnectionIdToAcknowledgements(final Ackno
             final ConnectionId connectionId) {
         final List<Acknowledgement> acksList = acknowledgements.stream()
                 .map(ack -> appendConnectionIdToAcknowledgementOrResponse(ack, connectionId))
-                .collect(Collectors.toList());
+                .toList();
         // Uses EntityId and StatusCode from input acknowledges expecting these were set when Acknowledgements was created
         return Acknowledgements.of(acknowledgements.getEntityId(), acksList, acknowledgements.getHttpStatus(),
                 acknowledgements.getDittoHeaders());
@@ -323,7 +323,7 @@ private Stream<SendingOrDropped> sendMappedOutboundSignal(final OutboundSignal.M
                                     outboundTargets);
                             return getSendingContextForTarget(outbound, target);
                         })
-                        .collect(Collectors.toList()));
+                        .toList());
 
 
         if (sendingContexts.isEmpty()) {

diff --git a/...rvice/src/main/java/org/eclipse/ditto/connectivity/service/messaging/ClientActorRefs.java b/...rvice/src/main/java/org/eclipse/ditto/connectivity/service/messaging/ClientActorRefs.java
@@ -17,7 +17,6 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.stream.Collectors;
 
 import javax.annotation.concurrent.NotThreadSafe;
 
@@ -89,7 +88,7 @@ public void clear() {
     public List<ActorRef> getOtherActors(final ActorRef clientActor) {
         return sortedRefs.stream()
                 .filter(actorRef -> !actorRef.equals(clientActor))
-                .collect(Collectors.toList());
+                .toList();
     }
 
     /**
@@ -130,7 +129,7 @@ public List<ActorRef> getSortedRefs() {
     }
 
     private static List<ActorRef> sort(final Map<ActorPath, ActorRef> refsByPath) {
-        return refsByPath.values().stream().sorted(ActorRef::compareTo).collect(Collectors.toList());
+        return refsByPath.values().stream().sorted(ActorRef::compareTo).toList();
     }
 
     @Override

diff --git a/...org/eclipse/ditto/connectivity/service/messaging/ExceptionToAcknowledgementConverter.java b/...org/eclipse/ditto/connectivity/service/messaging/ExceptionToAcknowledgementConverter.java
@@ -17,15 +17,15 @@
 import java.text.MessageFormat;
 import java.util.function.Predicate;
 
-import org.eclipse.ditto.json.JsonField;
-import org.eclipse.ditto.json.JsonObject;
-import org.eclipse.ditto.json.JsonPointer;
 import org.eclipse.ditto.base.model.acks.AcknowledgementLabel;
 import org.eclipse.ditto.base.model.common.HttpStatus;
 import org.eclipse.ditto.base.model.entity.id.EntityId;
 import org.eclipse.ditto.base.model.exceptions.DittoRuntimeException;
 import org.eclipse.ditto.base.model.headers.DittoHeaders;
 import org.eclipse.ditto.base.model.signals.acks.Acknowledgement;
+import org.eclipse.ditto.json.JsonField;
+import org.eclipse.ditto.json.JsonObject;
+import org.eclipse.ditto.json.JsonPointer;
 
 /**
  * Customizable converter from publisher exceptions to {@link Acknowledgement}s.
@@ -68,8 +68,8 @@ public final Acknowledgement convertException(final Exception exception,
         checkNotNull(dittoHeaders, "dittoHeaders");
 
         final Acknowledgement result;
-        if (exception instanceof DittoRuntimeException) {
-            result = convertDittoRuntimeException((DittoRuntimeException) exception, label, entityId, dittoHeaders);
+        if (exception instanceof DittoRuntimeException dittoRuntimeException) {
+            result = convertDittoRuntimeException(dittoRuntimeException, label, entityId, dittoHeaders);
         } else {
             result = convertGenericException(exception, label, entityId, dittoHeaders);
         }

diff --git a/...rc/main/java/org/eclipse/ditto/connectivity/service/messaging/InboundDispatchingSink.java b/...rc/main/java/org/eclipse/ditto/connectivity/service/messaging/InboundDispatchingSink.java
@@ -228,8 +228,8 @@ private Sink<Object, NotUsed> getSink() {
     }
 
     private static boolean isInboundMappingOutcomesWithError(final Object streamingElement) {
-        return streamingElement instanceof InboundMappingOutcomes &&
-                ((InboundMappingOutcomes) streamingElement).hasError();
+        return streamingElement instanceof InboundMappingOutcomes inboundMappingOutcomes &&
+                inboundMappingOutcomes.hasError();
     }
 
     private Sink<Object, NotUsed> onDittoRuntimeException() {
@@ -337,8 +337,7 @@ public Optional<Signal<?>> onError(final String mapperId,
         final Optional<Signal<?>> result;
         logger.debug("OnError mapperId=<{}> exception=<{}> topicPath=<{}> message=<{}>",
                 mapperId, e, topicPath, message);
-        if (e instanceof DittoRuntimeException) {
-            final var dittoRuntimeException = (DittoRuntimeException) e;
+        if (e instanceof DittoRuntimeException dittoRuntimeException) {
             if (isIllegalAdaptableException(dittoRuntimeException)) {
                 final var illegalAdaptableException = (IllegalAdaptableException) dittoRuntimeException;
                 if (isAboutInvalidLiveResponse(illegalAdaptableException)) {
@@ -871,7 +870,7 @@ private <T extends CommandResponse<T>> T appendConnectionIdToAcknowledgementOrRe
     private Acknowledgements appendConnectionIdToAcknowledgements(final Acknowledgements acknowledgements) {
         final List<Acknowledgement> acksList = acknowledgements.stream()
                 .map(this::appendConnectionIdToAcknowledgementOrResponse)
-                .collect(Collectors.toList());
+                .toList();
 
         // Uses EntityId and StatusCode from input acknowledges expecting these were set when Acknowledgements was created
         return Acknowledgements.of(acknowledgements.getEntityId(), acksList, acknowledgements.getHttpStatus(),

diff --git a/...rc/main/java/org/eclipse/ditto/connectivity/service/messaging/InboundMappingOutcomes.java b/...rc/main/java/org/eclipse/ditto/connectivity/service/messaging/InboundMappingOutcomes.java
@@ -133,8 +133,7 @@ public int hashCode() {
 
     @Override
     public boolean equals(final Object o) {
-        if (o instanceof InboundMappingOutcomes) {
-            final InboundMappingOutcomes that = (InboundMappingOutcomes) o;
+        if (o instanceof InboundMappingOutcomes that) {
             return Objects.equals(outcomes, that.outcomes) &&
                     Objects.equals(externalMessage, that.externalMessage) &&
                     Objects.equals(error, that.error) &&

diff --git a/...c/main/java/org/eclipse/ditto/connectivity/service/messaging/InboundMappingProcessor.java b/...c/main/java/org/eclipse/ditto/connectivity/service/messaging/InboundMappingProcessor.java
@@ -18,7 +18,6 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.function.Function;
-import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import javax.annotation.Nullable;
@@ -138,7 +137,7 @@ List<MappingOutcome<MappedInboundExternalMessage>> process(final ExternalMessage
                         (msg, header) -> msg.withHeader(header.getKey(), header.getValue()));
         return mappingTimer.overall(() -> mappers.stream()
                 .flatMap(mapper -> runMapper(mapper, externalMessageWithTraceContext, mappingTimer))
-                .collect(Collectors.toList())
+                .toList()
         );
     }
 
@@ -227,8 +226,8 @@ private static DittoRuntimeException toDittoRuntimeException(final Throwable err
                         e)
         );
 
-        if (error instanceof WithDittoHeaders) {
-            final DittoHeaders existingHeaders = ((WithDittoHeaders) error).getDittoHeaders();
+        if (error instanceof WithDittoHeaders withDittoHeaders) {
+            final DittoHeaders existingHeaders = withDittoHeaders.getDittoHeaders();
             final DittoHeaders mergedHeaders = bestEffortHeaders.toBuilder().putHeaders(existingHeaders)
                     .build();
             return dittoRuntimeException.setDittoHeaders(mergedHeaders);

diff --git a/.../service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/MappingTimer.java b/.../service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/MappingTimer.java
@@ -14,7 +14,6 @@
 
 import java.util.List;
 import java.util.function.Supplier;
-import java.util.stream.Collectors;
 
 import org.eclipse.ditto.base.model.signals.Signal;
 import org.eclipse.ditto.connectivity.api.ExternalMessage;
@@ -111,7 +110,7 @@ List<ExternalMessage> outboundPayload(final String mapper, final Supplier<List<E
         return timed(startedTimer, () -> {
             final List<ExternalMessage> externalMessages = supplier.get();
             return externalMessages.stream().map(em -> DittoTracing.propagateContext(context, em,
-                    (msg, entry) -> msg.withHeader(entry.getKey(), entry.getValue()))).collect(Collectors.toList());
+                    (msg, entry) -> msg.withHeader(entry.getKey(), entry.getValue()))).toList();
         });
     }
 
@@ -130,7 +129,7 @@ List<Adaptable> inboundPayload(final String mapper, final Supplier<List<Adaptabl
             final List<Adaptable> adaptables = supplier.get();
             return adaptables.stream()
                     .map(a -> DittoTracing.propagateContext(context, a))
-                    .collect(Collectors.toList());
+                    .toList();
         });
     }
 

diff --git a/.../main/java/org/eclipse/ditto/connectivity/service/messaging/OutboundDispatchingActor.java b/.../main/java/org/eclipse/ditto/connectivity/service/messaging/OutboundDispatchingActor.java
@@ -18,7 +18,6 @@
 import java.util.Set;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
-import java.util.stream.Collectors;
 
 import org.eclipse.ditto.base.model.acks.AcknowledgementLabel;
 import org.eclipse.ditto.base.model.acks.AcknowledgementLabelNotDeclaredException;
@@ -164,7 +163,7 @@ private Signal<?> adjustSignalAndStartAckForwarder(final Signal<?> signal, final
             return signal.setDittoHeaders(signal.getDittoHeaders().toBuilder()
                     .acknowledgementRequests(ackRequests.stream()
                             .filter(request -> targetIssuedAcks.contains(request.getLabel()))
-                            .collect(Collectors.toList()))
+                            .toList())
                     .build());
         }
     }

diff --git a/.../java/org/eclipse/ditto/connectivity/service/messaging/OutboundMappingProcessorActor.java b/.../java/org/eclipse/ditto/connectivity/service/messaging/OutboundMappingProcessorActor.java
@@ -444,7 +444,7 @@ private static Optional<JsonFieldSelector> getExtraFields(final ExpressionResolv
                         .map(expressionResolver::resolve)
                         .flatMap(PipelineElement::toStream)
                         .map(JsonPointer::of)
-                        .collect(Collectors.toList()))
+                        .toList())
                 .filter(jsonPointers -> !jsonPointers.isEmpty())
                 .map(JsonFactory::newFieldSelector)
                 .map(ThingFieldSelector::fromJsonFieldSelector);

diff --git a/...ity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/Resolvers.java b/...ity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/Resolvers.java
@@ -16,7 +16,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 import javax.annotation.Nullable;
 
@@ -69,7 +68,7 @@ private Resolvers() {
     private static final List<Placeholder<?>> PLACEHOLDERS = Collections.unmodifiableList(
             RESOLVER_CREATORS.stream()
                     .map(ResolverCreator::getPlaceholder)
-                    .collect(Collectors.toList())
+                    .toList()
     );
 
     /**

diff --git a/...rc/main/java/org/eclipse/ditto/connectivity/service/messaging/ResponseCollectorActor.java b/...rc/main/java/org/eclipse/ditto/connectivity/service/messaging/ResponseCollectorActor.java
@@ -15,7 +15,6 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.stream.Collectors;
 
 import org.eclipse.ditto.base.model.common.HttpStatus;
 import org.eclipse.ditto.base.model.exceptions.DittoRuntimeException;
@@ -160,7 +159,7 @@ public boolean allExpectedResponsesArrived() {
         public List<CommandResponse<?>> getFailedResponses() {
             return commandResponses.stream()
                     .filter(Output::isFailedResponse)
-                    .collect(Collectors.toList());
+                    .toList();
         }
 
         @Override
@@ -219,7 +218,7 @@ public int hashCode() {
 
         @Override
         public boolean equals(final Object other) {
-            return other instanceof SetCount && ((SetCount) other).count == count;
+            return other instanceof SetCount setCount && setCount.count == count;
         }
     }
 }