SpringBoot IntegrationFlowRegistration NPE when RuntimeException thrown from deadly handler

I'm using Spring Boot 2.2.7.RELEASE

I am getting a NPE when using IntegrationFlowRegistration for a File based Flow, with org.springframework.integration.dsl.IntegrationFlowDefinition.delay and when a file exists at startup in the Polling Directory. If the GenericHandler in https://docs.spring.io/spring-integration/api/org/springframework/integration/dsl/IntegrationFlowDefinition.html#handle-org.springframework.integration.handler.GenericHandler- throws a RuntimeException, then I get a NPE .

Source code that reproduces the error:

/**
 * Test the {@link org.springframework.integration.dsl#delay} with a file IntegrationFlowRegistration.
 */
@SpringBootTest
@ExtendWith(SpringExtension.class)
public class TestDelayFileIntegrationFlowRegistration {

  private static final Logger LOG = LoggerFactory.getLogger("TestDelayFileIntegrationFlowRegistration");

  static File pollDirectory2 = new File("/tmp/polltest2");


  /**
   * Configure the {@link IntegrationFlowRegistration}.
   */
  @Configuration
  @EnableIntegration
  public static class ContextConfiguration {


    private static File newFile2 = new File(pollDirectory2, "newFile2.txt");

    static {
      try {
        if (!newFile2.exists()) {
          newFile2.mkdirs();
        }
        newFile2.createNewFile();
      } catch (IOException e) {
        e.printStackTrace();
      }
    }

    @Autowired
    private IntegrationFlowContext flowContext;

    /**
     * Setup the {@link IntegrationFlowRegistration}.
     *
     * @return the {@link IntegrationFlowRegistration}.
     */
    @Bean
    public IntegrationFlowRegistration fileFlow2() {
      IntegrationFlow flow = IntegrationFlows
          .from(Files.inboundAdapter(pollDirectory2), e -> e.poller(Pollers.fixedDelay(100)))
          .delay("delayer",
              (DelayerEndpointSpec e) -> e.defaultDelay(1000))
          .handle(File.class, (file, headers) -> {
            LOG.info("**** Retry handling for " + file);
            if (file.exists()) {
              LOG.info("**** file exists");
              throw new RuntimeException();
            } else {
              LOG.info("*** No File");
            }
            return null;
          })
          .get();
      return this.flowContext.registration(flow).register();
    }

  }

  /**
   * Test the file flow.
   *
   * @throws Exception
   */
  @Test
  public void testFileFlow() throws Exception {

    (new File(pollDirectory2, "yetAnotherNewFile.txt")).createNewFile();

    Thread.sleep(5000);
  }

}

For the above code I'm seeing the following NPE:

20:43:34.468 [task-scheduler-4] ERROR o.s.i.handler.LoggingHandler - java.lang.NullPointerException at org.springframework.integration.handler.DelayHandler.releaseMessage(DelayHandler.java:454) at org.springframework.integration.handler.DelayHandler.lambda$releaseMessageAfterDelay$0(DelayHandler.java:412) at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at java.base/java.lang.Thread.run(Thread.java:834)

The following test that doesn't use an IntegrationFlowRegistration runs without throwing a NPE.

@SpringBootTest
@ExtendWith(SpringExtension.class)
public class TestDelayFileIntegrationFlowRegistration {

  private static final Logger LOG = LoggerFactory.getLogger("TestDelayFileIntegrationFlowRegistration");

  static File pollDirectory2 = new File("/tmp/polltest2");


  /**
   * Configure the {@link IntegrationFlowRegistration}.
   */
  @Configuration
  @EnableIntegration
  public static class ContextConfiguration {


    private static File newFile2 = new File(pollDirectory2, "newFile2.txt");

    static {
      try {
        if (!newFile2.exists()) {
          newFile2.mkdirs();
        }
        newFile2.createNewFile();
      } catch (IOException e) {
        e.printStackTrace();
      }
    }

    @Autowired
    private IntegrationFlowContext flowContext;

    /**
     * Setup the {@link IntegrationFlowRegistration}.
     *
     * @return the {@link IntegrationFlowRegistration}.
     */
    @Bean
    public IntegrationFlowRegistration fileFlow2() {
      IntegrationFlow flow = IntegrationFlows
          .from(Files.inboundAdapter(pollDirectory2), e -> e.poller(Pollers.fixedDelay(100)))
          .delay("delayer",
              (DelayerEndpointSpec e) -> e.defaultDelay(1000))
          .handle(File.class, (file, headers) -> {
            LOG.info("**** Retry handling for " + file);
            if (file.exists()) {
              LOG.info("**** file exists");
              throw new RuntimeException();

            } else {
              LOG.info("*** No `File");
            }
            return null;
          })
          .get();
      return this.flowContext.registration(flow).register();
    }

  }


  /**
   * Test the file flow.
   *
   * @throws Exception
   */
  @Test
  public void testFileFlow() throws Exception {

    (new File(pollDirectory2, "yetAnotherNewFile.txt")).createNewFile();

    Thread.sleep(5000);
  }

}

Comment From: snicoll

Thanks for the report but this a Spring Integration issue that should be reported to the relevant issue tracker. When you do, please consider moving all that code in text in an actual project that the team can run to reproduce the problem. Thank you.