Uploaded image for project: 'Spring Integration'
  1. Spring Integration
  2. INT-2993

JdbcChannelMessageStore thread safety issue with doPollForMessage and idCache

    XMLWordPrintable

    Details

      Description

      It is not easy to reproduce but every now and then I get an exception "Parameter "#??" is not set" like the below...

      I believe it is due to the fact that the Message Id's IN clause generation, so the MESSAGE_ID not in (?, ?, ?, ?, ?, ?, ?, ?, ?) and the binding to the actual parameter values is not atomic, so there can be a certain number of message id's in the list when we are generating the MESSAGE_ID not in (?, ?, ?, ?, ?, ?, ?, ?, ?), but there could have been message id's removed after this before we actually execute the statement. You will notice below that Parameter 11 is the last ? in the not in list. I would suggest that making a copy of the list in the synchronized (idCache) { would fix the issue.

      I know it's H2 and you don't support it, but I got this issue with HSQLDB as well, just not very often as it is not consistent because of the fact its a race condition.

      org.h2.jdbc.JdbcSQLException: Parameter "#11" is not set; SQL statement:
      SELECT INT_CHANNEL_MESSAGE.MESSAGE_ID, INT_CHANNEL_MESSAGE.MESSAGE_BYTES from INT_CHANNEL_MESSAGE where INT_CHANNEL_MESSAGE.GROUP_KEY = ? and INT_CHANNEL_MESSAGE.REGION = ? and INT_CHANNEL_MESSAGE.MESSAGE_ID not in (?, ?, ?, ?, ?, ?, ?, ?, ?) order by CREATED_DATE ASC LIMIT 1 [90012-161]
      at org.h2.message.DbException.getJdbcSQLException(DbException.java:329)
      at org.h2.message.DbException.get(DbException.java:169)
      at org.h2.message.DbException.get(DbException.java:146)
      at org.h2.expression.Parameter.checkSet(Parameter.java:73)
      at org.h2.command.Prepared.checkParameters(Prepared.java:162)
      at org.h2.command.CommandContainer.query(CommandContainer.java:81)
      at org.h2.command.Command.executeQuery(Command.java:185)
      at org.h2.jdbc.JdbcPreparedStatement.executeQuery(JdbcPreparedStatement.java:96)
      at org.springframework.jdbc.core.JdbcTemplate$1.doInPreparedStatement(JdbcTemplate.java:644)
      at org.springframework.jdbc.core.JdbcTemplate.execute(JdbcTemplate.java:587)
      at org.springframework.jdbc.core.JdbcTemplate.query(JdbcTemplate.java:637)
      at org.springframework.jdbc.core.JdbcTemplate.query(JdbcTemplate.java:662)
      at org.springframework.jdbc.core.JdbcTemplate.query(JdbcTemplate.java:702)
      at org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate.query(NamedParameterJdbcTemplate.java:166)
      at org.springframework.integration.jdbc.store.JdbcChannelMessageStore.doPollForMessage(JdbcChannelMessageStore.java:456)
      at org.springframework.integration.jdbc.store.JdbcChannelMessageStore.pollMessageFromGroup(JdbcChannelMessageStore.java:568)
      at org.springframework.integration.store.MessageGroupQueue.doPoll(MessageGroupQueue.java:278)
      at org.springframework.integration.store.MessageGroupQueue.poll(MessageGroupQueue.java:130)
      at org.springframework.integration.store.MessageGroupQueue.poll(MessageGroupQueue.java:47)
      at org.springframework.integration.channel.QueueChannel.doReceive(QueueChannel.java:91)
      at org.springframework.integration.channel.AbstractPollableChannel.receive(AbstractPollableChannel.java:57)
      at org.springframework.integration.endpoint.PollingConsumer.receiveMessage(PollingConsumer.java:79)
      at org.springframework.integration.endpoint.AbstractTransactionSynchronizingPollingEndpoint.doPoll(AbstractTransactionSynchronizingPollingEndpoint.java:67)
      at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:146)
      at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:144)
      at sun.reflect.GeneratedMethodAccessor25.invoke(Unknown Source)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
      at java.lang.reflect.Method.invoke(Method.java:597)
      at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:319)
      at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:183)
      at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:150)
      at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:110)
      at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172)
      at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:202)
      at $Proxy18.call(Unknown Source)
      at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller$1.run(AbstractPollingEndpoint.java:236)
      at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:52)
      at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
      at java.lang.Thread.run(Thread.java:662)
      04-16 15:44:52 jdbc[7]: exception

        Attachments

        1. JdbcChannelMessageStoreTest.java
          2 kB
        2. patch.txt
          1 kB
        3. patch.txt
          1 kB

          Activity

            People

            Assignee:
            abilan Artem Bilan
            Reporter:
            pellcorp Jason Pell
            Votes:
            0 Vote for this issue
            Watchers:
            3 Start watching this issue

              Dates

              Created:
              Updated:
              Resolved: