Adapted usage of placeholders to make use of multiple resolved values…

… where possible

* Extracting AuthorizationContext: use all resolved values for SubjectId generation
* SignalEnforcement: Use first resolved value for 'input', use all resolved values for 'filters'
* ConnectionStatusMessageMapper: Use first resolved value for featureId, use first resolved value for thingId of modifyFeatureProperty command
* ImplicitThingCreationMessageMapper: Use first resolved value for CreateThing JSON
* KafkaConsumerGroupSpecificConfig: Use first resolved value for consumer group ID
* KafkaValidator: Removed Source address placeholder validation (makes no sense)
* KafkaValidator: Use first resolved value for target address
* Abstract MqttValidator: Just validate that all placeholders are valid. Ignore results

connectivity/api/src/main/java/org/eclipse/ditto/connectivity/api/SignalEnforcementFilter.java

@@ -70,14 +70,16 @@ final class SignalEnforcementFilter implements EnforcementFilter<Signal<?>> {
     @Override
     public void match(final Signal<?> filterInput) {
         final EntityId entityId = extractEntityId(filterInput)
-                        .orElseThrow(() -> getEnforcementFailedException(filterInput.getDittoHeaders()));
+                .orElseThrow(() -> getEnforcementFailedException(filterInput.getDittoHeaders()));
         final List<UnresolvedPlaceholderException> exceptions = new LinkedList<>();
 
         for (final Placeholder<EntityId> filterPlaceholder : filterPlaceholders) {
             for (final String filter : enforcement.getFilters()) {
                 try {
-                    final String resolved = PlaceholderFilter.apply(filter, entityId, filterPlaceholder);
-                    if (inputValue.equals(resolved)) {
+                    final boolean anyResolvedToInputValue = PlaceholderFilter.applyForAll(filter, entityId, filterPlaceholder)
+                            .stream()
+                            .anyMatch(inputValue::equals);
+                    if (anyResolvedToInputValue) {
                         return;
                     }
                 } catch (final UnresolvedPlaceholderException unresolved) {
@@ -143,4 +145,5 @@ public String toString() {
                 ", inputValue=" + inputValue +
                 "]";
     }
+
 }

connectivity/api/src/test/java/org/eclipse/ditto/connectivity/api/placeholders/TopicPlaceholderTest.java

@@ -103,17 +103,17 @@ public void testPlaceholderFilter() {
         assertThatExceptionOfType(IllegalArgumentException.class).isThrownBy(
                 () -> topicPlaceholder.resolveValues(knownTopicPath, ""));
         assertThatExceptionOfType(UnresolvedPlaceholderException.class).isThrownBy(
-                () -> PlaceholderFilter.apply("{{ topic:unknown }}", knownTopicPath, topicPlaceholder));
+                () -> PlaceholderFilter.applyForAll("{{ topic:unknown }}", knownTopicPath, topicPlaceholder));
         assertThatExceptionOfType(UnresolvedPlaceholderException.class).isThrownBy(
-                () -> PlaceholderFilter.apply("{{ {{  topic:name  }} }}", knownTopicPath, topicPlaceholder));
-        assertThat(PlaceholderFilter.apply("eclipse:ditto", knownTopicPath, topicPlaceholder)).isEqualTo(
-                "eclipse:ditto");
-        assertThat(PlaceholderFilter.apply("prefix:{{ topic:channel }}:{{ topic:group }}:suffix", knownTopicPath,
-                topicPlaceholder)).isEqualTo("prefix:twin:things:suffix");
-
-        assertThat(PlaceholderFilter.apply("{{topic:subject}}", knownTopicPathSubject1,
-                topicPlaceholder)).isEqualTo(knownSubject);
-        assertThat(PlaceholderFilter.apply("{{  topic:action-subject}}", knownTopicPathSubject2,
-                topicPlaceholder)).isEqualTo(knownSubject2);
+                () -> PlaceholderFilter.applyForAll("{{ {{  topic:name  }} }}", knownTopicPath, topicPlaceholder));
+        assertThat(PlaceholderFilter.applyForAll("eclipse:ditto", knownTopicPath, topicPlaceholder))
+                .containsExactly("eclipse:ditto");
+        assertThat(PlaceholderFilter.applyForAll("prefix:{{ topic:channel }}:{{ topic:group }}:suffix", knownTopicPath,
+                topicPlaceholder)).containsExactly("prefix:twin:things:suffix");
+
+        assertThat(PlaceholderFilter.applyForAll("{{topic:subject}}", knownTopicPathSubject1,
+                topicPlaceholder)).containsExactly(knownSubject);
+        assertThat(PlaceholderFilter.applyForAll("{{  topic:action-subject}}", knownTopicPathSubject2,
+                topicPlaceholder)).containsExactly(knownSubject2);
     }
 }

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/mapping/ConnectionStatusMessageMapper.java

@@ -12,6 +12,8 @@
  */
 package org.eclipse.ditto.connectivity.service.mapping;
 
+import static org.eclipse.ditto.placeholders.PlaceholderFilter.apply;
+
 import java.time.Instant;
 import java.util.Collections;
 import java.util.List;
@@ -35,7 +37,6 @@
 import org.eclipse.ditto.placeholders.ExpressionResolver;
 import org.eclipse.ditto.placeholders.HeadersPlaceholder;
 import org.eclipse.ditto.placeholders.PlaceholderFactory;
-import org.eclipse.ditto.placeholders.PlaceholderFilter;
 import org.eclipse.ditto.protocol.Adaptable;
 import org.eclipse.ditto.protocol.adapter.DittoProtocolAdapter;
 import org.eclipse.ditto.things.model.Feature;
@@ -128,8 +129,8 @@ private List<Adaptable> doMap(final ExternalMessage externalMessage) {
         final Map<String, String> externalHeaders = externalMessage.getHeaders();
 
         final ExpressionResolver expressionResolver = getExpressionResolver(externalHeaders);
-        final ThingId thingId = ThingId.of(applyPlaceholderReplacement(mappingOptionThingId, expressionResolver));
-        final String featureId = applyPlaceholderReplacement(mappingOptionFeatureId, expressionResolver);
+        final ThingId thingId = ThingId.of(apply(mappingOptionThingId, expressionResolver));
+        final String featureId = apply(mappingOptionFeatureId, expressionResolver);
 
         //Check if time is convertible
         final long creationTime = extractLongHeader(externalHeaders, HEADER_HONO_CREATION_TIME,
@@ -215,10 +216,6 @@ private Adaptable getModifyFeatureAdaptable(final ThingId thingId, final String
         return DITTO_PROTOCOL_ADAPTER.toAdaptable(modifyFeature);
     }
 
-    private String applyPlaceholderReplacement(final String template, final ExpressionResolver expressionResolver) {
-        return PlaceholderFilter.apply(template, expressionResolver);
-    }
-
     private static ExpressionResolver getExpressionResolver(final Map<String, String> headers) {
         return PlaceholderFactory.newExpressionResolver(
                 PlaceholderFactory.newPlaceholderResolver(HEADERS_PLACEHOLDER, headers));

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/mapping/ImplicitThingCreationMessageMapper.java

@@ -174,7 +174,7 @@ public List<Adaptable> map(final ExternalMessage message) {
 
         final String resolvedTemplate;
         if (Placeholders.containsAnyPlaceholder(thingTemplate)) {
-            resolvedTemplate = applyPlaceholderReplacement(thingTemplate, expressionResolver);
+            resolvedTemplate = PlaceholderFilter.apply(thingTemplate, expressionResolver);
         } else {
             resolvedTemplate = thingTemplate;
         }
@@ -204,10 +204,6 @@ private static ExpressionResolver getHeadersExpressionResolver(final Map<String,
                 PlaceholderFactory.newPlaceholderResolver(HEADERS_PLACEHOLDER, headers));
     }
 
-    private static String applyPlaceholderReplacement(final String template, final ExpressionResolver resolver) {
-        return PlaceholderFilter.apply(template, resolver);
-    }
-
     private Signal<CreateThing> getCreateThingSignal(final ExternalMessage message, final String template) {
         final JsonObject thingJson = wrapJsonRuntimeException(() -> JsonFactory.newObject(template));
         final Thing newThing = ThingsModelFactory.newThing(thingJson);

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

@@ -38,6 +38,7 @@
 import org.eclipse.ditto.connectivity.service.messaging.Resolvers;
 import org.eclipse.ditto.connectivity.service.messaging.validation.AbstractProtocolValidator;
 import org.eclipse.ditto.placeholders.PlaceholderFactory;
+import org.eclipse.ditto.placeholders.UnresolvedPlaceholderException;
 
 import akka.actor.ActorSystem;
 
@@ -107,9 +108,7 @@ protected void validateSource(final Source source, final DittoHeaders dittoHeade
 
         final String placeholderReplacement = UUID.randomUUID().toString();
         source.getAddresses().forEach(address -> {
-            final String addressWithoutPlaceholders = validateTemplateAndReplace(address, dittoHeaders,
-                    placeholderReplacement, Resolvers.getPlaceholders());
-            validateSourceAddress(addressWithoutPlaceholders, dittoHeaders, placeholderReplacement);
+            validateSourceAddress(address, dittoHeaders, placeholderReplacement);
         });
 
         validateSourceQos(source, dittoHeaders);
@@ -121,7 +120,9 @@ protected void validateTarget(final Target target, final DittoHeaders dittoHeade
 
         final String placeholderReplacement = UUID.randomUUID().toString();
         final String addressWithoutPlaceholders = validateTemplateAndReplace(target.getAddress(), dittoHeaders,
-                placeholderReplacement, Resolvers.getPlaceholders());
+                placeholderReplacement, Resolvers.getPlaceholders()).stream()
+                .findFirst()
+                .orElseThrow(() -> UnresolvedPlaceholderException.newBuilder(target.getAddress()).build());
 
         validateTargetAddress(addressWithoutPlaceholders, dittoHeaders, placeholderReplacement);
         validateHeaderMapping(target.getHeaderMapping(), dittoHeaders);