Spring Integration Flow ClassCastException after Spring Boot 3.0.1 Update

Question

I was working on the Spring Boot Update to 3.0.1 and with it the Upgrade of Spring Integration 6.0.0 and Spring Cloud Stream 4.0.0. After upgrading however my previously working Spring Integration Flow is failing with underlying ClassCastException:

class org.springframework.messaging.handler.HandlerMethod$HandlerMethodParameter cannot be cast to class java.lang.reflect.Type (org.springframework.messaging.handler.HandlerMethod$HandlerMethodParameter is in unnamed module of loader 'app'; java.lang.reflect.Type is in module java.base of loader 'bootstrap')

Update: The exception originates from JsonMessageConverter inside of Spring Cloud Function Context, when trying to cast an object conversionHint (in my case of type HandlerMethodParameter) to Type.

Cannot cast 'org.springframework.messaging.handler.HandlerMethod$HandlerMethodParameter' to 'java.lang.reflect.Type'

Any hints or suggestions as to what might be the problem are highly appreciated.

Following is a highly shortened version of the affected integration flow and corresponding code snippets and a more detailed explanation:

@Bean
IntegrationFlow extract(SessionFactory<SftpClient.DirEntry> sftpSessionFactory,
                        XmlFileTransformer xmlFileTransformer){
   return IntegrationFlow
            .from(Sftp.inboundAdapter(sftpSessionFactory)
                    .preserveTimestamp(true)
                    .remoteDirectory("foo")
                    .regexFilter(".*\\.txt$")
                    .localDirectory(new File("sftp-inbound")), e -> e.id("sftpInboundAdapter")
                    .autoStartup(true)
                    .poller(Pollers.fixedDelay(5000))
            )
            .log(LoggingHandler.Level.DEBUG, "ExtractFlow", m -> "Successfully reached")
            .wireTap(MONITORING_FLOW)
            .log(LoggingHandler.Level.DEBUG, "ExtractFlow", m -> "Successfully done wire tap")
            .transform(xmlFileTransformer)
            .log(LoggingHandler.Level.DEBUG, "ExtractFlow", m -> "Successfully done transformation")
            .handle(m -> xmlProcessor.process((XmlFile) m.getPayload())
            .get();
}


@RequiredArgsConstructor
@Component
public class XmlFileTransformer implements GenericTransformer<Message<File>, XmlFile> {
   @Override
   public XmlFile transform(Message<File> message) {
       return new XmlFile(message.getPayload().toPath(), message.getHeaders().get("x-origin", String.class));
   }
}

The Integration Flow from the wiretap again transforms by implementing GenericTransfomer (similary written as XmlFileTransformer and then uses Amqp.outboundAdapter to send messages. The method xmlProcessor.process takes XmlFile as argument. However it never reaches the actual method because it breaks when trying to go through the wire tap, and in case I comment the wire tap it breaks when trying to transform with xmlFileTransformer. So after log message "Successfully reached" the exception happens.

I'm using following relevant dependencies (other used dependencies not listed for better overview):

<dependencyManagement>
 <dependencies>
     <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-dependencies</artifactId>
        <version>2022.0.0</version>
        <type>pom</type>
        <scope>import</scope>
     </dependency>
  </dependencies>
</dependencyManagement>

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.0.1</version>
    <relativePath/>
</parent>

<dependencies>
   <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-stream</artifactId>
   </dependency>
   <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
   </dependency>
   <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-starter</artifactId>
   </dependency>
   <dependency>
      <groupId>org.springframework.integration</groupId>
      <artifactId>spring-integration-sftp</artifactId>
   </dependency>
   <dependency>
      <groupId>org.springframework.integration</groupId>
      <artifactId>spring-integration-jdbc</artifactId>
   </dependency>
   <dependency>
      <groupId>org.springframework.integration</groupId>
      <artifactId>spring-integration-xml</artifactId>
   </dependency>
   <dependency>
     <groupId>org.springframework.integration</groupId>
      <artifactId>spring-integration-amqp</artifactId>
   </dependency>
</dependencies>

I was checking and rechecking my dependencies to ensure that I don't have any old dependencies that might clash with the update. However those should be fine as I'm using mostly the versions from spring-boot-starter-parent.

According to the migration guide of Spring Integration 6.0.0 there shouldn't be any major breaks that I forgot to handle. I was trying to find any information about similar cases however it seems not too many have either tried to upgrade to Spring Boot 3 yet or just didn't have the same issues as me. Could this be a bug in Spring Integration or do I need to refactor code so it's still running in Spring Integration 6.0? Or did I overlook issues with my dependencies?

Answer 1

I implemented a work around for the problem. I found out that the ApplicationJsonMessagingMarshallingConverter() was removed in Spring Cloud 4.0.x due to deprecation. This removal resulted in the fact, that every time when I was calling some kind of transform() in the IntegrationFlow the Message was converted with JsonMessageConverter where the actual ClassCassException is being thrown quite correctly.

Thinking I could just add the MappingJackson2MessageConverter explicitly I tried to add the bean manually, but the converter was not added to the set of "eligible" message converters to delegate to, as it was filtered out in ContentTypeConfiguration.java in cloud-stream-config package.

However the following workaround did the trick:

public class CustomMessageMarshallingConverter extends MappingJackson2MessageConverter {}

Then registering this "custom" converter as bean:

@Bean
public MessageConverter customMessageConverter(){
   return new CustomMessageMarshallingConverter();
}

The original MappingJackson2MessageConverter was added to the available message converters and correctly identified that in fact no conversion was necessary at this point.

This feels like a hack, however I'm just glad to have working code.

Answer 2

Disclaimer, I was not updating to Spring Boot 3, but rather writing a new Spring Boot 3 app that uses Cloud Stream. The other services that communicate to this new service are all still in Spring Boot 2.3

For me (using Spring boot 3.0.2) the error was, ironically, caused by how the message was sent to the RMQ. On the producer side, I was sending the messages through standard RabbitTemplate and not Spring's StreamBridge.

While I am not entirely familiar with Spring Cloud Steam's message decoding feature, RabbitTemplate in standard configuration is sending a base64 encoded Java object including its package name.

This also persists if RabbitTemplate is set to use JSON, i.e.

var eventPayload = new SomeEventDto(12332, "some string val", "another string val");
template.setMessageConverter(new Jackson2JsonMessageConverter());
    template.convertAndSend("rabbit-stream-exchange", "rabbit-stream-queue", eventPayload); 

Sending the message from the abovementioned code adds additional header to the rabbit message:

__TypeId__: com.example.example123.SomeEventDto

If the Spring Cloud Stream receiver receives the message with __TypeId__ header set it will then fail to cast the message into the correct object, seemingly even if the header is set to the exact type that it is supposed to deserialize into. At the same time, if the same payload is (manually) sent without this header, all other parameters unchanged, the message will be consumed successfully.

Thus, my fix was to send the message through the Spring Cloud Stream instead:

@Service
@RequiredArgsConstructor
public class EventPublisher {
  private final StreamBridge streamBridge;

  public void sendSomeEvent(SomeEventDto event) throws IOException {
    streamBridge.send("some-event-binding-out-0", MessageBuilder.withPayload(event).build());
  }
}

The message sent this way was serialized as JSON and sent without the `__TypeId__` header which was confusing the consumer so much.