Spring Integration Implementation Review / Feeds Stop Processing Randomly

Question

I have inherited an application with a lot of Spring Integration components (and I am not a very familiar with spring integration). It is an eCommerce application and has integrations for Products, Stock, Prices etc.

The problem is most common with the (Large) ProductFeed which comes into the system in almost 2000 individual JMS messages which are grouped (correlationId) and resequenced in the correct order after being retrieved from an Apache MQ server. This is done so dependent objects are saved into the DB before others. A consequence of the structure of the export from the Product Information system. But it has also been witnessed with other feeds such as prices.

We have 3 admin servers / JVMs with the same spring configuration. There is an issue where sometimes the Product Feeds will randomly stop processing the messages after 1 or 2 messages of the 2000 messages in the feed. The messages all enqueue successfully in AMQ, then the queue just seem to drain all the messages but only 1 or 2 actually log they are received and processed.

This only seems to happen in environments with more than one admin server. In the spring integration documentation I cannot see how multiple JVMs can function in a cluster to communicate on which messages to retrieve, and I am wondering if somehow the JMS Consumers are conflicting with each other causing this problem (as these are large feeds with lots of messages and live on the queues for a long time before being retrieved and resequenced).

Has anyone encountered this feed processing issues in the past?
Is there a better architecture we should be using to process messages like this or are we missing something from my spring integration configuration.

I have abbreviated some of the spring code to keep it to the spring integration code, products specifically and not all the processing beans. All help gratefully received.

<bean id="jmsDefaultRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
    <property name="queue" value="*"/>
    <property name="maximumRedeliveries" value="${import.jms.redelivery.count}"/>
    <property name="initialRedeliveryDelay" value="${import.jms.redelivery.delay}"/>
</bean>

<bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="${import.jms.url}"/>
    <property name="redeliveryPolicy" ref="jmsDefaultRedeliveryPolicy"/>
</bean>

<integration:control-bus input-channel="operationChannel"/>

<!-- message resequencing flow -->
<integration:channel id="messageRoutingInputChannel"/>

<integration:header-value-router input-channel="messageRoutingInputChannel" header-name="REQUIRES_RESEQUENCING">
    <integration:mapping value="true" channel="messageResequenceInputChannel" />
    <integration:mapping value="false" channel="messageHandleInputChannel" />
</integration:header-value-router>

<integration:channel id="messageResequenceInputChannel"/>

<integration:chain input-channel="messageResequenceInputChannel" output-channel="messageHandleInputChannel">
    <integration:header-enricher id="messageHeaderEnricher">
        <integration:header name="correlationId" expression="headers['jms_correlationId']" />
        <integration:header name="sequenceNumber" expression="headers['JMSXGroupSeq']" />
        <integration:header name="sequenceSize" expression="headers['GROUP_SIZE']" />
    </integration:header-enricher>

    <integration:resequencer id="messageResequencer" release-partial-sequences="true" />
</integration:chain>
<!-- message resequencing flow -->

<integration:channel id="messageHandleInputChannel"/>

<integration:header-value-router input-channel="messageHandleInputChannel" header-name="FEED_TYPE">
    <integration:mapping value="Assets" channel="messageProductInputChannel"/>
    <integration:mapping value="Attributes" channel="messageProductInputChannel"/>
    <integration:mapping value="DeletedCategories" channel="messageProductInputChannel"/>
    <integration:mapping value="Categories" channel="messageProductInputChannel"/>
    <integration:mapping value="VariantCategories" channel="messageProductInputChannel"/>
    <integration:mapping value="DeletedBaseProducts" channel="messageProductInputChannel"/>
    <integration:mapping value="BaseProducts" channel="messageProductInputChannel"/>
    <integration:mapping value="Products" channel="messageProductInputChannel"/>
    <integration:mapping value="ProductCrossReferences" channel="messageProductInputChannel"/>
</integration:header-value-router>

<bean id="baseMessageHandler" abstract="true" class="com.mycompany.dataimport.service.feed.impl.AbstractMessageHandler">
    <property name="activeTenantId" value="${tenantId}" />
</bean>

<util:list id="activeChannelAdapters">
    <value>productMessageAdapter</value>
</util:list>

<bean id="baseFeedImportStrategy" abstract="true" class="com.mycompany.dataimport.service.feed.impl.AbstractFeedImportStrategy">
    <property name="feedSessionSetups">
        <list>
            <ref bean="adminUserSessionSetup"/>
            <ref bean="catalogSessionSetup"/>
        </list>
    </property>
    <property name="errorLoggers">
        <list>
            <bean class="com.mycompany.dataimport.service.feed.impl.ConsoleImportErrorLogger" />
            <ref bean="emailImportErrorLogger" />
        </list>
    </property>
</bean>

<bean id="productQueue" class="org.apache.activemq.command.ActiveMQQueue">
    <constructor-arg index="0" value="${import.queue.feed.product}" />
</bean>

<jms:message-driven-channel-adapter id="productMessageAdapter" channel="messageRoutingInputChannel" destination="productQueue" connection-factory="jmsConnectionFactory"
    acknowledge="transacted" auto-startup="false" concurrent-consumers="${concurrent.consumers.product.queue}" max-concurrent-consumers="${max.concurrent.consumers.product.queue}" />

<integration:channel id="messageProductInputChannel" />

<integration:service-activator id="productMessageActivator" input-channel="messageProductInputChannel" ref="productMessageHandler" />

<bean id="productMessageHandler" parent="baseMessageHandler" class="com.mycompany.dataimport.service.feed.product.ProductMessageHandler" scope="prototype">
    <property name="importStrategyMap">
        <map>
            <entry key="Assets" value-ref="assetFeedImportStrategy" />
            <entry key="Attributes" value-ref="attributeFeedImportStrategyChain" />
            <entry key="VariantCategories" value-ref="variantCategoryFeedImportStrategy" />
            <entry key="Categories" value-ref="categoryFeedImportStrategyChain" />
            <entry key="DeletedCategories" value-ref="deleteCategoryFeedImportStrategy" />
            <entry key="DeletedBaseProducts" value-ref="disableBaseProductImportStrategy" />
            <entry key="BaseProducts" value-ref="baseProductFeedImportStrategy" />
            <entry key="Products" value-ref="productFeedImportStrategyChain" />
            <entry key="ProductCrossReferences" value-ref="productReferenceFeedImportStrategy" />
        </map>
    </property>
</bean>

Gary Russell Gary Russell · Accepted Answer · 2017-02-01T13:43:28

Since your instances are all listening to the same queue, they will compete for messages. ActiveMQ has a prefetch configuration, which is used to improve performance (rather than fetching the messages one-at-a-time). It defaults to 1000 so it is possible that, since you only have 2000 messages, some of your consumer(s) might be starved for messages. Especially if you have multiple consumers in each container, which your configuration implies.

You can try reducing the prefetch to see if it helps distribute the work better across your consumers.

If that doesn't work, in my experience, such "hangs" are usually due to the listener thread being "stuck" in user code. You should take a thread dump to see what the listener container threads are doing.

Spring Integration Implementation Review / Feeds Stop Processing Randomly

1 Answers