IllegalArgumentException when using --mode=ref

[sabbyanandan]

As a user, I have a stream that includes file source with --mode=ref and on every new file event, I'm getting an IllegalArgumentException. This is happening on both Avogadro-GA (1.1.0.GA) and Avogadro-SR1 (1.1.1.GA) releases. This, however, works with 1.0.2.GA release, though.

stream:

stream create foo --definition "file --filename-pattern=*.jpg --directory=/data --mode=ref | log"

error:

2017-01-29 16:03:24.985 DEBUG 9088 --- [ask-scheduler-1] o.s.i.e.SourcePollingChannelAdapter      : Poll resulted in Message: GenericMessage [payload=/data/salsa.jpg, headers={id=2d361087-8d6e-5161-e935-30e84c6803ec, timestamp=1485734604985}]
2017-01-29 16:03:24.985 DEBUG 9088 --- [ask-scheduler-1] o.s.integration.channel.DirectChannel    : preSend on channel 'output', message: GenericMessage [payload=/data/salsa.jpg, headers={id=2d361087-8d6e-5161-e935-30e84c6803ec, timestamp=1485734604985}]
2017-01-29 16:03:24.985 DEBUG 9088 --- [ask-scheduler-1] o.s.i.channel.PublishSubscribeChannel    : preSend on channel 'errorChannel', message: ErrorMessage [payload=org.springframework.messaging.MessageDeliveryException: failed to send Message to channel 'output'; nested exception is java.lang.IllegalArgumentException: payload must not be null, headers={id=9900c562-5037-7551-feb7-a5ad9f31aba4, timestamp=1485734604985}]
2017-01-29 16:03:24.985 DEBUG 9088 --- [ask-scheduler-1] o.s.integration.handler.LoggingHandler   : _org.springframework.integration.errorLogger.handler received message: ErrorMessage [payload=org.springframework.messaging.MessageDeliveryException: failed to send Message to channel 'output'; nested exception is java.lang.IllegalArgumentException: payload must not be null, headers={id=9900c562-5037-7551-feb7-a5ad9f31aba4, timestamp=1485734604985}]
2017-01-29 16:03:24.986 ERROR 9088 --- [ask-scheduler-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageDeliveryException: failed to send Message to channel 'output'; nested exception is java.lang.IllegalArgumentException: payload must not be null
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:449)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373)
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)
	at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105)
	at org.springframework.integration.endpoint.SourcePollingChannelAdapter.handleMessage(SourcePollingChannelAdapter.java:203)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:272)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint.access$000(AbstractPollingEndpoint.java:58)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:190)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:186)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller$1.run(AbstractPollingEndpoint.java:353)
	at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:55)
	at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
	at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:51)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller.run(AbstractPollingEndpoint.java:344)
	at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
	at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:81)
	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:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.IllegalArgumentException: payload must not be null
	at org.springframework.util.Assert.notNull(Assert.java:115)
	at org.springframework.integration.support.MutableMessage.<init>(MutableMessage.java:57)
	at org.springframework.integration.support.MutableMessage.<init>(MutableMessage.java:53)
	at org.springframework.integration.support.MutableMessageBuilder.withPayload(MutableMessageBuilder.java:86)
	at org.springframework.integration.support.MutableMessageBuilderFactory.withPayload(MutableMessageBuilderFactory.java:35)
	at org.springframework.integration.support.MutableMessageBuilderFactory.withPayload(MutableMessageBuilderFactory.java:26)
	at org.springframework.cloud.stream.binding.MessageConverterConfigurer$ContentTypeConvertingInterceptor.preSend(MessageConverterConfigurer.java:194)
	at org.springframework.integration.channel.AbstractMessageChannel$ChannelInterceptorList.preSend(AbstractMessageChannel.java:538)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:415)
	... 23 more

[artembilan]

@sabbyanandan ,

The MessageConverterConfigurer has a logic like:

final BindingProperties bindingProperties = this.channelBindingServiceProperties.getBindingProperties(
				channelName);
final String contentType = bindingProperties.getContentType();
if (!input && bindingProperties.getProducer() != null && bindingProperties.getProducer().isPartitioned()) {
	messageChannel.addInterceptor(new PartitioningInterceptor(bindingProperties));
}
if (StringUtils.hasText(contentType)) {
	messageChannel.addInterceptor(new ContentTypeConvertingInterceptor(contentType, input));
}

So, if there is some contentType property configured for the binding that looks like it tries to convert the File payload to the expected contentType and fails with the null.

So, share, please, which binder properties you have for the file source.

[sabbyanandan]

Thanks for looking into this, @artembilan. I'm using rabbit-app with all the default configs. The latest rabbit-apps can be found here.

[artembilan]

M-m-m. That doesn't help.
Sorry.
Look, I did this in the test-case for mode=ref:

"spring.cloud.stream.bindings.output.contentType=text/plain"

And voila! I ended up with the same IllegalArgumentException: payload must not be null, just because StringMessageConverter can't convert File object to the String via:

if (this.provideHint) {
	converted = ((AbstractMessageConverter) this.messageConverter).toMessage(message.getPayload(),
			new MutableMessageHeaders(message.getHeaders()), this.mimeType);
}

That must be fixed, but on the SCSt side.
I assume the simple fallback to the message.getPayload() should be enough in the end if we can't convert it to the expected contentType.

The file source isn't guilty.

[garyrussell]

This is a known issue with the MessageConverterConfigurer - if message conversion fails, it tries to create a message with a null payload; it should detect the converter returns null and throw a more meaningful error.

We added code to SI to permit suppressing the propagation of an inbound contentType to the outbound side; this looks like another flavor of that.

It seems to me that the "canConvert" test should be based on the payload type, not the contentType header (or fall back to the payload type if it finds the contentType is not convertible).

@mbogoevici ??

[artembilan]

@garyrussell ,

???

It is file source: there is no inbound headers.
See my comment: #2 (comment). The contentType comes from the binder configuration.
We select the target converter based on that option.
And than we fail here because File doesn't meet StringMessageConverter requirements:

@Override
protected boolean supports(Class<?> clazz) {
	return (String.class == clazz);
}

[garyrussell]

I was just pointing out the flaw in the converter; I wasn't commenting on the origin of the contentType that was used for the conversion.

That's what I meant by "another flavor".

[mbogoevici]

As per our IPM discussion, this is a SCSt regression. Looks like we have lost the ability to do simple Object->String conversions. StringMessageConverter won't work because it's specifically focused on String to byte[] (internal/wire format) conversions with a given encoding (i.e. when the attached contentType is text/plain).

My suggestion would be to:

a) as per @garyrussell 's suggestion, throw an exception when conversion fails instead of returning null

b) Include a unidirectional ObjectToString converter that can handle any non-String to outbound String (via payload.toString()) - it would be unidirectional, because there's no generic way to convert an inbound String to an POJO. cc/@artembilan

[mbogoevici]

Basically this is what I think that we're missing: https://github.com/spring-cloud/spring-cloud-stream/blob/1.0.x/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToStringMessageConverter.java

Possibly a more enhanced version that can use encoding.

[artembilan]

The fix in SCSt has been merged.
Local testing confirms it.
This issue should be addressed with an additional test-case when an appropriate SCSt version is picked up by app-starters-code.