log after split messes up reactive output channel check

[asarkar]

Reference: my own question on stackoverflow.
https://stackoverflow.com/q/47725873/839733

It appears that the log statement uses wiretap to change the output channel of the splitter to a DirectChannel which messes up the following logic in AbstractMessageSplitter.
boolean reactive = getOutputChannel() instanceof ReactiveStreamsSubscribableChannel;

[artembilan]

What is the problem though you would like to see here fixed?

That is how it works. That's all. If you want the reactive output, you should explicitly specify it. The log() is definitely wire-tap for the current MessageChannel, but since there is no channel after split(), it inserts one and by default we use DirectChannel.

Maybe some Docs clarification would be enough?

WDYT?

[asarkar]

There is a channel in the end, although not immediately after split (see code below where the channel used to be immediately before get). If the log channel can be delayed until the whole flow has been resolved, that will be great. Otherwise, at least, the splitter section should warn about this pitfall. As far as I saw, the splitter is the only component that has a reactive channel check.

return IntegrationFlows.from(sourceProperties.getArtifactStreamingInputChannel())
.filter(Message.class, msg -> msg.getHeaders().containsKey(FILE_TYPE_HEADER))
.handle(new GzipToFluxTransformer(artifactoryProperties))
.transform((Flux<String> payload) -> payload
        .onBackpressureBuffer(sourceProperties.getOnBackpressureBufferSize(),
                s -> log.error("Buffer overrun!")))
.split()
.channel(c -> c.flux(sourceProperties.getArtifactStreamingOutputChannel()))
.log(LoggingHandler.Level.DEBUG, CLASS_NAME, Message::getHeaders)
.get();

[artembilan]

??? You can place as much log() in the flow definition as you wish and in any place. It is applied only to the channel where it is declared. So, if you place it directly after split() only splitter results will be logged.
That clearly states in the Doc that log() is tied to the current MessageChannel in the flow, not the next one. There fore if you would like to log from the FluxMessageChannel, you definitely should specify it before log() operator. Not sure what made you think differently...

[asarkar]

Not sure what made you think differently...

Perhaps my lack of experience working with Spring integration. If this is the expected behavior, that's ok, but a warning note in the doc would be useful. For large data stream, it really makes a difference and I can see other people tripping on this. For small number of messages, it doesn't matter so even if someone makes the same mistake as I did, they'd not realize it.

[artembilan]

Sure! I'll do an IMPORTANT note in the mention doc for log()!

[artembilan]

@asarkar ,

I don't know if GH notifies about linked PRs, but I'll take a responsibility and let you know that here it is #2305

[asarkar]

@artembilan Appreciate it. If you make a commit and reference to an issue, it shows up here but I don't get notified.
I looked at the PR and have a suggestion - perhaps you should mention that for large volume of data, the automatic creation of a DirectChannel may be a performance problem. In my case, it certainly was.

[artembilan]

Well, sorry. That's not related to the integration flow definition. The DirectChannel does nothing about data, just propagate it into the subscriber in the same thread.

That is already details of the channel implementation and not how log() is applied.