@Test
public void testSignalEnrichmentWithPayloadMappedTargets() {

    resetActorSystemWithCachingSignalEnrichmentProvider();
    final TestProbe proxyActorProbe = TestProbe.apply("mockConciergeForwarder", actorSystem);
    setUpProxyActor(proxyActorProbe.ref());
    new TestKit(actorSystem) {{
        // GIVEN: MessageMappingProcessor started with a test probe as the configured enrichment provider
        final ActorRef underTest = createMessageMappingProcessorActor(this);
        // WHEN: a signal is received with 6 targets:
        //  - 1 with enrichment w/o payload mapping
        //  - 1 with enrichment with 1 payload mapping
        //  - 1 with enrichment with 2 payload mappings
        //  - 1 w/o enrichment w/o payload mapping
        //  - 1 w/o enrichment with 1 payload mapping
        //  - 1 w/o enrichment with 2 payload mappings
        final JsonFieldSelector extraFields = JsonFactory.newFieldSelector("attributes/x,attributes/y",
                JsonParseOptions.newBuilder().withoutUrlDecoding().build());
        final AuthorizationSubject targetAuthSubject = AuthorizationSubject.newInstance("target:auth-subject");
        final AuthorizationSubject targetAuthSubjectWithoutIssuer =
                AuthorizationSubject.newInstance("auth-subject");
        final Target targetWithEnrichment = ConnectivityModelFactory.newTargetBuilder()
                .address("target/address")
                .authorizationContext(AuthorizationContext.newInstance(DittoAuthorizationContextType.UNSPECIFIED,
                        targetAuthSubject))
                .topics(ConnectivityModelFactory.newFilteredTopicBuilder(Topic.TWIN_EVENTS)
                        .withExtraFields(extraFields)
                        .build())
                .build();
        final Target targetWithEnrichmentAnd1PayloadMapper = ConnectivityModelFactory.newTargetBuilder()
                .address("target/address/mapped/1")
                .authorizationContext(AuthorizationContext.newInstance(DittoAuthorizationContextType.UNSPECIFIED,
                        targetAuthSubject))
                .topics(ConnectivityModelFactory.newFilteredTopicBuilder(Topic.TWIN_EVENTS)
                        .withExtraFields(extraFields)
                        .build())
                .payloadMapping(ConnectivityModelFactory.newPayloadMapping(ADD_HEADER_MAPPER))
                .build();
        final Target targetWithEnrichmentAnd2PayloadMappers = ConnectivityModelFactory.newTargetBuilder()
                .address("target/address/mapped/2")
                .authorizationContext(AuthorizationContext.newInstance(DittoAuthorizationContextType.UNSPECIFIED,
                        targetAuthSubject))
                .topics(ConnectivityModelFactory.newFilteredTopicBuilder(Topic.TWIN_EVENTS)
                        .withExtraFields(extraFields)
                        .build())
                .payloadMapping(ConnectivityModelFactory.newPayloadMapping(DUPLICATING_MAPPER, ADD_HEADER_MAPPER))
                .build();
        final Target targetWithoutEnrichment = ConnectivityModelFactory.newTargetBuilder(targetWithEnrichment)
                .address("target/address/without/enrichment")
                .topics(Topic.TWIN_EVENTS)
                .build();
        final Target targetWithoutEnrichmentAnd1PayloadMapper =
                ConnectivityModelFactory.newTargetBuilder(targetWithEnrichment)
                        .address("target/address/without/enrichment/with/1/payloadmapper")
                        .topics(Topic.TWIN_EVENTS)
                        .payloadMapping(ConnectivityModelFactory.newPayloadMapping(ADD_HEADER_MAPPER))
                        .build();
        final Target targetWithoutEnrichmentAnd2PayloadMappers =
                ConnectivityModelFactory.newTargetBuilder(targetWithEnrichment)
                        .address("target/address/without/enrichment/with/2/payloadmappers")
                        .topics(Topic.TWIN_EVENTS)
                        .payloadMapping(
                                ConnectivityModelFactory.newPayloadMapping(ADD_HEADER_MAPPER, DUPLICATING_MAPPER))
                        .build();
        final Signal<?> signal = TestConstants.thingModified(Collections.emptyList())
                .setRevision(8L); // important to set revision to same value as cache lookup retrieves
        final OutboundSignal outboundSignal = OutboundSignalFactory.newOutboundSignal(signal, Arrays.asList(
                targetWithEnrichment,
                targetWithoutEnrichment,
                targetWithEnrichmentAnd1PayloadMapper,
                targetWithoutEnrichmentAnd1PayloadMapper,
                targetWithEnrichmentAnd2PayloadMappers,
                targetWithoutEnrichmentAnd2PayloadMappers)
        );
        underTest.tell(outboundSignal, getRef());
        // THEN: Receive a RetrieveThing command from the facade.
        final RetrieveThing retrieveThing = proxyActorProbe.expectMsgClass(RetrieveThing.class);
        final JsonFieldSelector extraFieldsWithAdditionalCachingSelectedOnes = JsonFactory.newFieldSelectorBuilder()
                .addPointers(extraFields)
                .addFieldDefinition(Thing.JsonFields.REVISION) // additionally always select the revision
                .build();
        assertThat(retrieveThing.getSelectedFields()).contains(extraFieldsWithAdditionalCachingSelectedOnes);
        assertThat(retrieveThing.getDittoHeaders().getAuthorizationContext()).containsExactly(targetAuthSubject,
                targetAuthSubjectWithoutIssuer);
        final JsonObject extra = JsonObject.newBuilder()
                .set("/attributes/x", 5)
                .build();
        final JsonObject extraForCachingFacade = JsonObject.newBuilder()
                .set("_revision", 8)
                .setAll(extra)
                .build();
        proxyActorProbe.reply(RetrieveThingResponse.of(retrieveThing.getEntityId(), extraForCachingFacade,
                retrieveThing.getDittoHeaders()));
        // THEN: mapped messages arrive in a batch.
        final PublishMappedMessage publishMappedMessage = expectMsgClass(PublishMappedMessage.class);
        int i = 0;
        // THEN: the first mapped signal is without enrichment
        expectPublishedMappedMessage(publishMappedMessage, i++, signal, targetWithoutEnrichment,
                mapped -> assertThat(mapped.getExternalMessage().getHeaders()).containsOnlyKeys("content-type")
        );
        // THEN: the second mapped signal is without enrichment and applied 1 payload mapper arrives
        expectPublishedMappedMessage(publishMappedMessage, i++, signal,
                targetWithoutEnrichmentAnd1PayloadMapper,
                mapped -> assertThat(mapped.getExternalMessage().getHeaders())
                        .contains(AddHeaderMessageMapper.OUTBOUND_HEADER)
        );
        // THEN: a mapped signal without enrichment and applied 2 payload mappers arrives causing 3 messages
        //  as 1 mapper duplicates the message
        expectPublishedMappedMessage(publishMappedMessage, i++, signal,
                targetWithoutEnrichmentAnd2PayloadMappers,
                mapped -> assertThat(mapped.getExternalMessage().getHeaders())
                        .contains(AddHeaderMessageMapper.OUTBOUND_HEADER)
        );
        expectPublishedMappedMessage(publishMappedMessage, i++, signal,
                targetWithoutEnrichmentAnd2PayloadMappers,
                mapped -> assertThat(mapped.getExternalMessage().getHeaders()).containsOnlyKeys("content-type")
        );
        expectPublishedMappedMessage(publishMappedMessage, i++, signal,
                targetWithoutEnrichmentAnd2PayloadMappers,
                mapped -> assertThat(mapped.getExternalMessage().getHeaders()).containsOnlyKeys("content-type")
        );
        // THEN: Receive an outbound signal with extra fields.
        expectPublishedMappedMessage(publishMappedMessage, i++, signal, targetWithEnrichment,
                mapped -> assertThat(mapped.getAdaptable().getPayload().getExtra()).contains(extra),
                mapped -> assertThat(mapped.getExternalMessage().getHeaders()).containsOnlyKeys("content-type")
        );
        // THEN: Receive an outbound signal with extra fields and with mapped payload.
        expectPublishedMappedMessage(publishMappedMessage, i++, signal,
                targetWithEnrichmentAnd1PayloadMapper,
                mapped -> assertThat(mapped.getAdaptable().getPayload().getExtra()).contains(extra),
                mapped -> assertThat(mapped.getExternalMessage().getHeaders())
                        .contains(AddHeaderMessageMapper.OUTBOUND_HEADER)
        );
        // THEN: a mapped signal with enrichment and applied 2 payload mappers arrives causing 3 messages
        //  as 1 mapper duplicates the message
        expectPublishedMappedMessage(publishMappedMessage, i++, signal,
                targetWithEnrichmentAnd2PayloadMappers,
                mapped -> assertThat(mapped.getAdaptable().getPayload().getExtra()).contains(extra),
                mapped -> assertThat(mapped.getExternalMessage().getHeaders()).containsOnlyKeys("content-type")
        );
        expectPublishedMappedMessage(publishMappedMessage, i++, signal,
                targetWithEnrichmentAnd2PayloadMappers,
                mapped -> assertThat(mapped.getAdaptable().getPayload().getExtra()).contains(extra),
                mapped -> assertThat(mapped.getExternalMessage().getHeaders()).containsOnlyKeys("content-type")
        );
        expectPublishedMappedMessage(publishMappedMessage, i, signal,
                targetWithEnrichmentAnd2PayloadMappers,
                mapped -> assertThat(mapped.getAdaptable().getPayload().getExtra()).contains(extra),
                mapped -> assertThat(mapped.getExternalMessage().getHeaders())
                        .contains(AddHeaderMessageMapper.OUTBOUND_HEADER)
        );
    }};
}