spring-cloud-stream: Nullpointer exception when return null message in functional streaming

In short when i return null using functional programming model instead of filtering message i get java.lang.NullPointerException: null

This is what we were using until now:

@EnableBinding(Processor.class)
public class ClassName {

    @StreamListener(Sink.INPUT)
    @SendTo(OUTPUT)
    public Message<OutputClass> process(@Payload InputClass input) {
        if (isMessageWrong) {
            // this the way we were filtering messages
            log.info(FILTERED_UNSUPPORTED_MESSAGE, key);
            return null;
        }
        return MessageBuilder
                    .withPayload(buildPayload(input))
                    .setHeader(Constants.MESSAGE_KEY_OUTGOING, key)
                    .build();
}

After upgrade to spring boot 2.4.2 and upgrade spring cloud stream to 3.1.0 is this aproach depreacted: * @deprecated as of 3.1 in favor of functional programming model

so we rework code to this, using functional programming model:

@Component
public class ClassName implements Function<Message<InputClass>, Message<OutputClass>> {

    @Override
    public Message<OutputClass> apply(@Payload InputClass input) {
        if (isMessageWrong) {
            // this the way we are trying to filter messages
            log.info(FILTERED_UNSUPPORTED_MESSAGE, key);
            return null;
        }
        return MessageBuilder
                    .withPayload(buildPayload(input))
                    .setHeader(Constants.MESSAGE_KEY_OUTGOING, key)
                    .build();
}

using this code, when i filter any wrong message and return null i’ve got this stacktrace:

Caused by: java.lang.NullPointerException: null
	at org.springframework.messaging.core.GenericMessagingTemplate.sendTimeout(GenericMessagingTemplate.java:252) ~[spring-messaging-5.3.3.jar:5.3.3]
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166) ~[spring-messaging-5.3.3.jar:5.3.3]
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47) ~[spring-messaging-5.3.3.jar:5.3.3]
	at org.springframework.messaging.core.AbstractDestinationResolvingMessagingTemplate.send(AbstractDestinationResolvingMessagingTemplate.java:72) ~[spring-messaging-5.3.3.jar:5.3.3]
	at org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder$1.doSendMessage(FunctionConfiguration.java:585) ~[spring-cloud-stream-3.1.0.jar:3.1.0]
	at org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder$1.handleMessageInternal(FunctionConfiguration.java:571) ~[spring-cloud-stream-3.1.0.jar:3.1.0]
	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:56) ~[spring-integration-core-5.4.3.jar:5.4.3]

the whole problem in my opinion is starting in org.springframework.cloud.stream.function.FunctionConfiguration.FunctionToDestinationBinder#createFunctionHandler

@Override
public void handleMessageInternal(Message<?> message) throws MessagingException {
	Object result = functionInvocationWrapper.apply((Message<byte[]>) message);
	if (result instanceof Iterable) {
		for (Object resultElement : (Iterable<?>) result) {
			this.doSendMessage(resultElement, message);
		}
	}
	else if (ObjectUtils.isArray(result) && !(result instanceof byte[])) {
		for (int i = 0; i < ((Object[]) result).length; i++) {
			this.doSendMessage(((Object[]) result)[i], message);
		}
	}
	else {
		this.doSendMessage(result, message);
	}
}

Why there is not any null check? Is there any prediction that users should not return null from thier apply methods? If yes, how am i supposed to skip wrong message?

About this issue

Original URL
State: closed
Created 3 years ago
Comments: 23 (10 by maintainers)

Commits related to this issue

GH-2107 Fix NPE for cases when functioin returns null Resolves #2107 — committed to olegz/spring-cloud-stream by olegz 3 years ago
GH-2107 Add test to validate no NPE for function returning null — committed to olegz/spring-cloud-stream by olegz 3 years ago

Most upvoted comments

Thanks for reporting this. Will definitely address in the next release and we’ll back-port it as well. By February 11 we should have a service release with the fix. Meanwhile you can still use the old approach for now.

olegz on Feb 4, 2021

I’ve ran it against current master and it passes just fine. Here is the updated POM from your app

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.4.2</version>
    </parent>

    <modelVersion>4.0.0</modelVersion>

    <artifactId>dummy-app</artifactId>
    <version>1.0.0.BUILD-SNAPSHOT</version>

    <properties>
        <java.version>11</java.version>
        <spring-cloud.version>2020.0.1</spring-cloud.version>
        <spring-cloud-stream.version>3.1.0</spring-cloud-stream.version>
    </properties>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-stream-dependencies</artifactId>
                <version>${spring-cloud-stream.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream-binder-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream</artifactId>
             <version>3.1.2-SNAPSHOT</version>
        </dependency>


        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-tomcat</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-undertow</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <mainClass>com.dummy.app.DummyAppApplication</mainClass>
                    <finalName>dummy-app</finalName>
                    <skip>false</skip>
                    <addResources>true</addResources>
                </configuration>
                <executions>
                    <execution>
                        <goals>
                            <goal>build-info</goal>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

If you have troubles getting the right snapshot in simply clone the project

git clone git@github.com:spring-cloud/spring-cloud-stream.git
cd spring-cloud-stream
./mvnw clean install -DskipTests

olegz on Feb 10, 2021