Uploaded image for project: 'Spring XD'
  1. Spring XD
  2. XD-1958

filejdbc job broken in distributed mode

    XMLWordPrintable

    Details

    • Type: Bug
    • Status: Done
    • Priority: Critical
    • Resolution: Complete
    • Affects Version/s: None
    • Fix Version/s: 1.0 RC1
    • Component/s: Batch
    • Labels:
    • Story Points:
      5
    • Rank (Obsolete):
      44962
    • Sprint:
      Sprint 31

      Description

      The filejdbc job is broken in distributed mode (redis and rabbit)

      To reproduce:

      export XD_TRANSPORT=rabbit

      start xd-admin
      start xd-container

      start shell and create this job:

      >job create mydata --definition "filejdbc --names=col1,col2,col3 --resources=file:///home/trisberg/Test/input/*.csv --initializeDatabase=true" --deploy
      >job launch mydata
      

      results in JOB starting but never completing:

      >job execution list
        Id  Job Name  Start Time                            Step Execution Count  Execution Status  Deployment Status  Definition Status
        --  --------  ------------------------------------  --------------------  ----------------  -----------------  -----------------
          0  mydata      2014-07-11 15:44:33 America/New_York  0                     STARTED           Deployed           Exists
      

      Steps:

      Step Id	Step Name	Reads	Writes	Commits	Rollbacks	Duration	Status	Details
      0	step1-master	0	0	0	0	-1405349644032 ms	EXECUTING	
      1	step1-master:partition0	292	292	3	0	302 ms	COMPLETED	
      2	step1-master:partition1	292	292	3	0	203 ms	COMPLETED	
      3	step1-master:partition2	292	292	3	0	193 ms	COMPLETED	
      

      When using Redis, I also get this stacktrace in container:

      15:40:51,220  INFO DeploymentsPathChildrenCache-0 boot.SpringApplication - Started application in 1.965 seconds (JVM running for 66.949)
      15:40:51,220  INFO DeploymentsPathChildrenCache-0 core.SimpleModule - initialized module: SimpleModule [name=filejdbc, type=job, group=job1, index=0 @64a28a58]
      15:40:51,233  INFO DeploymentsPathChildrenCache-0 redis.RedisMessageBus - binding requestor: job1.0
      15:40:51,236  INFO DeploymentsPathChildrenCache-0 redis.RedisMessageBus - binding replier: job1.0
      15:40:51,243  INFO DeploymentsPathChildrenCache-0 module.ModuleDeployer - deployed SimpleModule [name=filejdbc, type=job, group=job1, index=0 @64a28a58]
      15:40:57,110 ERROR inbound.job1.0-redis:queue-inbound-channel-adapter1 redis.RedisMessageBus$1 - Failed to deliver message; retries exhausted; message sent to queue 'ERRORS:name'
      org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.integration.aggregator.AggregatingMessageHandler#0]
      	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:84)
      	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
      	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      	at java.lang.reflect.Method.invoke(Method.java:606)
      	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:317)
      	at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190)
      	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
      	at org.springframework.integration.monitor.SimpleMessageHandlerMetrics.handleMessage(SimpleMessageHandlerMetrics.java:106)
      	at org.springframework.integration.monitor.SimpleMessageHandlerMetrics.invoke(SimpleMessageHandlerMetrics.java:86)
      	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
      	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:207)
      	at com.sun.proxy.$Proxy84.handleMessage(Unknown Source)
      	at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
      	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:101)
      	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:97)
      	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
      	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:255)
      	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:223)
      	at sun.reflect.GeneratedMethodAccessor89.invoke(Unknown Source)
      	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      	at java.lang.reflect.Method.invoke(Method.java:606)
      	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:317)
      	at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190)
      	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
      	at org.springframework.integration.monitor.DirectChannelMetrics.monitorSend(DirectChannelMetrics.java:113)
      	at org.springframework.integration.monitor.DirectChannelMetrics.doInvoke(DirectChannelMetrics.java:97)
      	at org.springframework.integration.monitor.DirectChannelMetrics.invoke(DirectChannelMetrics.java:91)
      	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
      	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:207)
      	at com.sun.proxy.$Proxy78.send(Unknown Source)
      	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:109)
      	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:44)
      	at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:94)
      	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.sendMessage(AbstractReplyProducingMessageHandler.java:260)
      	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.sendReplyMessage(AbstractReplyProducingMessageHandler.java:241)
      	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.produceReply(AbstractReplyProducingMessageHandler.java:205)
      	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleResult(AbstractReplyProducingMessageHandler.java:199)
      	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:177)
      	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78)
      	at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
      	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:101)
      	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:97)
      	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
      	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:255)
      	at org.springframework.xd.dirt.integration.redis.RedisMessageBus$1$1.doWithRetry(RedisMessageBus.java:251)
      	at org.springframework.xd.dirt.integration.redis.RedisMessageBus$1$1.doWithRetry(RedisMessageBus.java:247)
      	at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:263)
      	at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:168)
      	at org.springframework.xd.dirt.integration.redis.RedisMessageBus$1.doSend(RedisMessageBus.java:247)
      	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:255)
      	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:223)
      	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:109)
      	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:44)
      	at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:94)
      	at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:98)
      	at org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint.popMessageAndSend(RedisQueueMessageDrivenEndpoint.java:211)
      	at org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint.access$300(RedisQueueMessageDrivenEndpoint.java:50)
      	at org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint$ListenerTask.run(RedisQueueMessageDrivenEndpoint.java:290)
      	at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:52)
      	at java.lang.Thread.run(Thread.java:744)
      Caused by: java.lang.IllegalStateException: Null correlation not allowed.  Maybe the CorrelationStrategy is failing?
      	at org.springframework.util.Assert.state(Assert.java:385)
      	at org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler.handleMessageInternal(AbstractCorrelatingMessageHandler.java:383)
      	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78)
      	... 60 more
      15:41:00,129 ERROR inbound.job1.0-redis:queue-inbound-channel-adapter1 redis.RedisMessageBus$1 - Failed to deliver message; retries exhausted; message sent to queue 'ERRORS:name'
      org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.integration.aggregator.AggregatingMessageHandler#0]
      	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:84)
      	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
      	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      	at java.lang.reflect.Method.invoke(Method.java:606)
      	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:317)
      	at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190)
      	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
      	at org.springframework.integration.monitor.SimpleMessageHandlerMetrics.handleMessage(SimpleMessageHandlerMetrics.java:106)
      	at org.springframework.integration.monitor.SimpleMessageHandlerMetrics.invoke(SimpleMessageHandlerMetrics.java:86)
      	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
      	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:207)
      	at com.sun.proxy.$Proxy84.handleMessage(Unknown Source)
      	at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
      	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:101)
      	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:97)
      	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
      	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:255)
      	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:223)
      	at sun.reflect.GeneratedMethodAccessor89.invoke(Unknown Source)
      	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      	at java.lang.reflect.Method.invoke(Method.java:606)
      	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:317)
      	at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190)
      	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
      	at org.springframework.integration.monitor.DirectChannelMetrics.monitorSend(DirectChannelMetrics.java:113)
      	at org.springframework.integration.monitor.DirectChannelMetrics.doInvoke(DirectChannelMetrics.java:97)
      	at org.springframework.integration.monitor.DirectChannelMetrics.invoke(DirectChannelMetrics.java:91)
      	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
      	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:207)
      	at com.sun.proxy.$Proxy78.send(Unknown Source)
      	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:109)
      	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:44)
      	at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:94)
      	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.sendMessage(AbstractReplyProducingMessageHandler.java:260)
      	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.sendReplyMessage(AbstractReplyProducingMessageHandler.java:241)
      	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.produceReply(AbstractReplyProducingMessageHandler.java:205)
      	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleResult(AbstractReplyProducingMessageHandler.java:199)
      	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:177)
      	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78)
      	at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
      	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:101)
      	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:97)
      	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
      	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:255)
      	at org.springframework.xd.dirt.integration.redis.RedisMessageBus$1$1.doWithRetry(RedisMessageBus.java:251)
      	at org.springframework.xd.dirt.integration.redis.RedisMessageBus$1$1.doWithRetry(RedisMessageBus.java:247)
      	at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:263)
      	at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:168)
      	at org.springframework.xd.dirt.integration.redis.RedisMessageBus$1.doSend(RedisMessageBus.java:247)
      	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:255)
      	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:223)
      	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:109)
      	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:44)
      	at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:94)
      	at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:98)
      	at org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint.popMessageAndSend(RedisQueueMessageDrivenEndpoint.java:211)
      	at org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint.access$300(RedisQueueMessageDrivenEndpoint.java:50)
      	at org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint$ListenerTask.run(RedisQueueMessageDrivenEndpoint.java:290)
      	at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:52)
      	at java.lang.Thread.run(Thread.java:744)
      Caused by: java.lang.IllegalStateException: Null correlation not allowed.  Maybe the CorrelationStrategy is failing?
      	at org.springframework.util.Assert.state(Assert.java:385)
      	at org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler.handleMessageInternal(AbstractCorrelatingMessageHandler.java:383)
      	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78)
      	... 60 more
      

        Attachments

          Activity

            People

            Assignee:
            grussell Gary Russell
            Reporter:
            thomas.risberg Thomas Risberg
            Votes:
            0 Vote for this issue
            Watchers:
            2 Start watching this issue

              Dates

              Created:
              Updated:
              Resolved: