Need some guidance with Spring Integration Flow

Question

I am new to Spring Integration and have read quite some documentation and other topics here on StackOverflow. But I am still a bit overwhelmed on how to apply the newly acquired knowledge in a Spring Boot Application. This is what should happen:

1. receive message from a Kafka topic, eg from "request-topic" (payload is a custom Job POJO). InboundChannelAdapter?
2. do some preparation (checkout from a git repo)
3. process files using a batch job
4. commit&push to git, update Job object with commit-id
5. publish message to Kafka with updated Job object, eg to "reply-topic". OutboundChannelAdapter?

Using DSL or plain Java configuration does not matter. My problem after trying several variants is that I could not achieve the desired result. For example, handlers would be called too early, or not at all, and thus the reply in step 5 would not be updated. Also, there should only be one flow running at any given time, so I guess, a queue should be involved at some point, probably at step 1(?).

Where and when should I use QueueChannels, DirectChannel (or any other?), do I need GatewayHandlers, eg to reply with a commit-id? Any hints are appreciated.

Answer 1

Something like this:

@Bean
IntegrationFlow flow() {
    return IntegrationFlows.from(Kafka.inboundGateway(...))
            .handle(// prep)
            .transform(// to JobLaunchRequest)
            .handle(// JobLaunchingGateway)
            .handle(// cleanUp and return result)
            .get();
}

It will only process one request at a time (with default concurrency).