spring http events spring-integration enterprise-integration

Spring Integration HTTP to Scatter Gather

I'm new to Spring Integration and trying to make use of the enterprise pattern of scatter-gather, but I'm struggling with implementation details and struggling with available examples I can find online.

In short my scenario is:

An HTTP request is sent from a user to system A.
Before responding (aka synchronous), system A sends N number of messages to N number of system Xs, asynchronously.
System A sits and waits for responses.
Once there is a response from each of the request systems, system A collates the responses into one larger response.
System A finally responds to the user with the larger response.

Basically, as far as the original consumer is concerned, a single is request that responds with an answer, without having to 'come back later'. However, that request was actually to a facade that masks the complexity that lies behind it (potentially hitting hundreds of systems, making synchronous requests at back-end non-performant and infeasible).

So far I have this implementation (scrubbed details so may not be 1:1 example of what I'm playing with, for example the correlationStrategy I've since worked out isn't doing what I'd expect):

@Bean
public IntegrationFlow overallRequest(final AmqpTemplate amqpTemplate) {

  return IntegrationFlows.from( // HTTP endpoint to user makes requests on
          Http.inboundChannelAdapter("/request-overall-document")
              .requestMapping(m -> m.methods(HttpMethod.POST))
              .requestPayloadType(String.class))
      .log()
      // Arbitrary header to simplify example, realistically would generate a UUID
      // and attach to some correlating header that works for systems involved
      .enrichHeaders(p -> p.header("someHeader", "someValue"))
      .log()
      .scatterGather(
          recipientListRouterSpec ->
              recipientListRouterSpec
                  .applySequence(true)
                  .recipientFlow(
                      flow ->
                          flow.handle( // Straight pass through of msg received to see in response
                              Amqp.outboundAdapter(amqpTemplate)
                                  .exchangeName( // RabbitMQ fanout exchange to N queues to N systems
                                      "request-overall-document-exchange"))),
          aggregatorSpec ->
              aggregatorSpec
                  // Again for example, arbitrary once two correlated responses
                  .correlationStrategy(msg -> msg.getHeaders().get("someHeader"))
                  .releaseStrategy(gm -> gm.size() == 2)
                  // Simple string concatenation for overall response
                  .outputProcessor(
                      msgrp ->
                          msgrp.getMessages().stream()
                              .map(msg -> msg.getPayload().toString())
                              .reduce("Overall response: ", (nexus, txt) -> nexus + "|" + txt))
                  // Reset group on each response
                  .expireGroupsUponCompletion(true),
          scatterGatherSpec ->
              scatterGatherSpec.gatherChannel(
                  responseChannel())) // The channel to listen for responses to request on
      .log()
      .get();
}

With this as the response channel configuration:

@Bean
public MessageChannel responseChannel() {
  return new QueueChannel();
}

@Bean
public AmqpInboundChannelAdapter responseChannelAdapter(
    SimpleMessageListenerContainer listenerContainer,
    @Qualifier("responseChannel") MessageChannel channel) {
  AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer);
  adapter.setOutputChannel(channel);
  return adapter;
}

@Bean
public SimpleMessageListenerContainer responseContainer(ConnectionFactory connectionFactory) {
  SimpleMessageListenerContainer container =
      new SimpleMessageListenerContainer(connectionFactory);
  container.setQueueNames("request-overall-document-responses");
  return container;
}

With all responses being sent to a separate Spring application that just pipes the request payloads back again (aka for testing without having to integrate with actual systems):

@Bean
public IntegrationFlow systemOneReception(final ConnectionFactory connectionFactory, final AmqpTemplate amqpTemplate) {
  return IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "request-overall-document-system-1"))
      .log()
      .handle(Amqp.outboundAdapter(amqpTemplate).routingKey("request-overall-document-responses"))
      .get();
}

@Bean
public IntegrationFlow systemTwoReception(final ConnectionFactory connectionFactory, final AmqpTemplate amqpTemplate) {
  return IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "request-overall-document-system-2"))
      .log()
      .handle(Amqp.outboundAdapter(amqpTemplate).routingKey("request-overall-document-responses"))
      .get();
}

And I get the following error in system A upon successful release as per the aggregation / release strategy in the scatter-gather implementation:

2020-02-29 20:06:39.255 ERROR 152 --- [ask-scheduler-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageDeliveryException: The 'gatherResultChannel' header is required to deliver the gather result., failedMessage=GenericMessage [payload=Overall response: |somerequesttobesent|somerequesttobesent, headers={amqp_receivedDeliveryMode=PERSISTENT, content-length=19, amqp_deliveryTag=2, sequenceSize=1, amqp_redelivered=false, amqp_contentEncoding=UTF-8, host=localhost:18081, someHeader=someValue, connection=keep-alive, correlationId=182ee203-85ab-9ef6-7b19-3a8e2da8f5a7, id=994a0cf5-ad2b-02c3-dc93-74fae2f5092b, cache-control=no-cache, contentType=text/plain, timestamp=1583006799252, http_requestMethod=POST, sequenceNumber=1, amqp_consumerQueue=request-overall-document-responses, accept=*/*, amqp_receivedRoutingKey=request-overall-document-responses, amqp_timestamp=Sat Feb 29 20:06:39 GMT 2020, amqp_messageId=3341deae-7ed0-a042-0bb7-d2d2be871165, http_requestUrl=http://localhost:18081/request-overall-document, amqp_consumerTag=amq.ctag-ULxwuAjp8ZzcopBZYvcbZQ, accept-encoding=gzip, deflate, br, user-agent=PostmanRuntime/7.22.0}]
at org.springframework.integration.scattergather.ScatterGatherHandler.lambda$doInit$2(ScatterGatherHandler.java:160)
at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:77)
at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:71)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:431)
at org.springframework.integration.handler.AbstractMessageProducingHandler.doProduceOutput(AbstractMessageProducingHandler.java:284)
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:265)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:223)
at org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler.completeGroup(AbstractCorrelatingMessageHandler.java:823)
at org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler.handleMessageInternal(AbstractCorrelatingMessageHandler.java:475)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:169)
at org.springframework.integration.endpoint.PollingConsumer.handleMessage(PollingConsumer.java:143)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:390)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:329)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$null$1(AbstractPollingEndpoint.java:277)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$2(AbstractPollingEndpoint.java:274)
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:93)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

Now I understand I have a few gaps, but I'm struggling to work out how to move forward:

The given error: that there isn't some 'gatherResultChannel' output. I'd have thought this would have been subsequent 'handles' / 'logs' / w.e on the result of the scatterGather(...) call, but not so on playing around.
There needs to be some form of mapping from the result of the scatter-gather aggregation back to the original Http.XXX request.

EDIT: from further digging, the issue given seems to be because when going out via AMQP (in my case, RabbitMQ), the header in question is deliberately dropped as it's a MessageChannel (see lines 230 to 257). Unsure if the implication here is that splitting/aggregation isn't intended to cross between multiple independent applications (my assumption is that it's dropped because it's an instance of a Java object, which would be problematic to pass around)...

FURTHER EDIT: with fresh eyes noticed something I hadn't before, the exception I pasted in quotes the failed message, and it seems to be a clear result of the output processing (while fiddling, flicked between DirectChannel and QueueChannel, only DirectChannel does not print the payload so wasn't looking for it). To be sure it wasn't doing some cloning or something weird, updated the stub service to transform and append unique postfixes (as below), and yes it was actually aggregating.

   .transform(msg -> MessageFormat.format("{0}_system1response", msg))
   .transform(msg -> MessageFormat.format("{0}_system2response", msg))

The 'gatherResultChannel' header is required to deliver the gather result., failedMessage=GenericMessage [payload=Overall response: |sometext_system2response|sometext_system1response, hea...

So it seems like scattering, gathering and aggregation is all working, the only thing that's not is that the given processing doesn't know where to push the messages after that?

ONCE MORE: As per Gary's response, replaced all adapters with gateways, however in doing so can no longer fanout? So removed scatterGatherSpec argument from the scatterGather call, and replaced / added in two recipient as follows:

.recipientFlow(flow -> flow.handle(Amqp.asyncOutboundGateway(asyncTemplate).routingKeyFunction(m -> "request-overall-document-system-1"), e -> e.id("sytemOneOutboundGateway")))
.recipientFlow(flow -> flow.handle(Amqp.asyncOutboundGateway(asyncTemplate).routingKeyFunction(m -> "request-overall-document-system-2"), e -> e.id("sytemTwoOutboundGateway")))

which is the closest I can get to a working example, however, while this does sort-of work, it results in reprocessing of message multiple times on/off queues, where my expected output for a POST with 'msgtosend' would have been:

Overall message: |msgtosend_system1response|msgtosend_system2response

Instead I get sporadic outputs like:

Overall message: |msgtosend|msgtosend_system1response
Overall message: |msgtosend_system2response|msgtosend_system1response_system1response
Overall message: |msgtosend|msgtosend_system1response_system1response
Overall message: |msgtosend_system2response|msgtosend_system1response_system1response

I assume there's some config / bean overlap but try as I might I can't isolate what it is, i.e. connection factory, listener container, async template, etc. etc.

Solution

Use an AMQP outbound gateway instead of outbound and inbound channel adapters; that way the channel header will be retained. There is an AsyncAmqpOutboundGateway which is probably best for your purposes.

If you must use channel adapters for some reason, use a header enricher together with a Header Channel Registry to convert the channel to a String representation so it can be retained.

EDIT

Here is a simple example:

@SpringBootApplication
public class So60469260Application {

    public static void main(String[] args) {
        SpringApplication.run(So60469260Application.class, args);
    }

    @Bean
    public IntegrationFlow flow(AsyncRabbitTemplate aTemp) {
        return IntegrationFlows.from(Gate.class)
                .enrichHeaders(he -> he.headerExpression("corr", "payload"))
                .scatterGather(rlr -> rlr
                        .applySequence(true)
                        .recipientFlow(f1 -> f1.handle(Amqp.asyncOutboundGateway(aTemp)
                                .routingKey("foo")))
                        .recipientFlow(f2 -> f2.handle(Amqp.asyncOutboundGateway(aTemp)
                                .routingKey("bar"))),
                        agg -> agg.correlationStrategy(msg -> msg.getHeaders().get("corr")))
                .get();
    }

    @Bean
    public AsyncRabbitTemplate aTemp(RabbitTemplate template) {
        return new AsyncRabbitTemplate(template);
    }

    @Bean
    @DependsOn("flow")
    public ApplicationRunner runner(Gate gate) {
        return args -> System.out.println(gate.doIt("foo"));
    }

    @RabbitListener(queues = "foo")
    public String foo(String in) {
        return in.toUpperCase();
    }

    @RabbitListener(queues = "bar")
    public String bar(String in) {
        return in + in;
    }

}

interface Gate {

    List<String> doIt(String in);

}

[foofoo, FOO]