[ROCKETMQ-102] When shutdown(), the persisted offet is not the latest consumed message, which may cause repeated messages. #64

Jaskey · 2017-02-16T05:33:28Z

Solution: add interface for push consumer to accept await termination time to await consuming.

JIRA: https://issues.apache.org/jira/browse/ROCKETMQ-102

UnitTest in DefaultMQPushConsumerTest.java

@Test
public void testShutdownAwait() throws Exception {
    final LinkedList<Long> consumedOffset = new LinkedList<>();
    pushConsumer.setPullInterval(0);
    pushConsumer.getDefaultMQPushConsumerImpl().setConsumeMessageService(new ConsumeMessageConcurrentlyService(pushConsumer.getDefaultMQPushConsumerImpl(), new MessageListenerConcurrently() {
        @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            for (MessageExt msg : msgs) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {e.printStackTrace();}
                synchronized (consumedOffset) {
                    consumedOffset.add(msg.getQueueOffset());
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }));
    pushConsumer.getDefaultMQPushConsumerImpl().doRebalance();
    PullMessageService pullMessageService = mQClientFactory.getPullMessageService();
    pullMessageService.executePullRequestImmediately(createPullRequest());
    Thread.sleep(1000);
    pushConsumer.shutdown(10 * 1000);//await consume for at most 10 seconds
    //pushConsumer.shutdown();//here if we do not await, the test case will not pass
    long persitOffset =pushConsumer.getDefaultMQPushConsumerImpl().getOffsetStore().readOffset(new MessageQueue(topic, brokerName, 0), ReadOffsetType.READ_FROM_MEMORY);
    Thread.sleep(1000);//wait for thread pool to continue consume for sometime if not terminated well
    Collections.sort(consumedOffset);
    Assert.assertEquals(consumedOffset.getLast() + 1, persitOffset);//when shutdown with await, the persisted offset should be the latest message offset
}

lizhanhui · 2017-02-16T09:25:01Z

Can you provide some test data, say before/after applying this patch, how many duplications are found respectively?

IMHO, we should make the API as concise as possible.

Jaskey · 2017-02-16T09:41:31Z

@lizhanhui

The problem exists for long, please review an old issue.
https://github.com/alibaba/RocketMQ/issues/367

The number of duplicated messages are depending on how many messages are being consumed in the thread pool or in the pending queue. This should be very easy to meet when consumptions are takes long and with massive accumulation.

The problem should be addressed since rocketmq should prevend duplication as much as possible.

The reason I add a new interface is that I want to make it compatible with the previous version, actually, if one shutdown(long awaitTerminations) is enough.

Or, if we don't mind, we could use a field to make this in push consumer, if we do this, no more interface is needed but a new field is added.

lizhanhui · 2017-02-16T09:49:05Z

I know this issue has been brought up in the past. What I suggest here is providing some testing data to consolidate the rationality of this patch. By doing so, you'll find it easier to get developers convinced.

Anyway, thanks a lot for bringing this issue back to attention.

Jaskey · 2017-02-16T12:35:31Z

@lizhanhui please refer to the test case and just switch the shutdown(long milis) method to shutdown() to find out the problem

Jaskey · 2017-02-16T12:39:31Z

@lizhanhui @zhouxinyu @vintagewang

How do you think my proposol of adding a field called awaitMilisWhenShutdown in DefaultPushConsumer, which will not need to add a shutdown(long awaitMilis) interface but one more config field.

coveralls · 2017-02-16T18:38:04Z

Coverage decreased (-0.3%) to 31.215% when pulling db9e92a on Jaskey:ROCKETMQ-102-shutdown-await into 573b22c on apache:master.

coveralls · 2017-02-17T00:52:23Z

Coverage decreased (-0.3%) to 31.215% when pulling db9e92a on Jaskey:ROCKETMQ-102-shutdown-await into 573b22c on apache:master.

coveralls · 2017-02-17T00:52:23Z

Coverage decreased (-0.3%) to 31.215% when pulling db9e92a on Jaskey:ROCKETMQ-102-shutdown-await into 573b22c on apache:master.

zhouxinyu · 2017-02-17T09:30:30Z

...rc/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java

+        if (awaitTerminateMillis > 0) {
+            try {
+                this.consumeExecutor.awaitTermination(awaitTerminateMillis,TimeUnit.MILLISECONDS);
+                if (!this.consumeExecutor.isTerminated()) log.info("There are messages still being consumed in thread pool, but not going to await them anymore. Have awaited for {} ms",awaitTerminateMillis);


Did you follow our code code guidelines[1] ? Below if block is recommended.

if { }

http://rocketmq.incubator.apache.org/docs/code-guidelines/

Ok, I will update the pr accordingly.

zhouxinyu · 2017-02-17T09:38:09Z

client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java

+        shutdown(0);
+    }
+
+    public void shutdown(long awaitTerminateMillis) {


Since MQPushConsumer only has the method interface for shutdown( ), so we could consider pull shutdown(long awaitTerminateMillis) up to the parent interface or make shutdown(long awaitTerminateMillis) private and give a default input(non-zero) in shutdown().

@zhouxinyu

I know and agree to your concern, actually, as I my previous comment, maybe we could add an configuration for awaitMills, what do you think ? The default value could be 0 to keep the same behavior

Good idea, it's ok for me.

zhouxinyu · 2017-02-17T09:39:14Z

Hi @Jaskey ,
Thanks for this PR, I added two comments for your reference.

coveralls · 2017-02-17T16:56:35Z

Coverage increased (+0.5%) to 31.992% when pulling b178d2f on Jaskey:ROCKETMQ-102-shutdown-await into 573b22c on apache:master.

coveralls · 2017-02-17T16:56:35Z

Coverage increased (+0.5%) to 31.992% when pulling b178d2f on Jaskey:ROCKETMQ-102-shutdown-await into 573b22c on apache:master.

Jaskey · 2017-02-20T02:14:15Z

@lizhanhui @zhouxinyu

please review the updated pr, which remains the same interface of push consumer.

coveralls · 2017-02-20T02:29:58Z

Coverage increased (+0.2%) to 31.691% when pulling d714c6c on Jaskey:ROCKETMQ-102-shutdown-await into 573b22c on apache:master.

coveralls · 2017-02-20T02:29:58Z

Coverage increased (+0.2%) to 31.691% when pulling d714c6c on Jaskey:ROCKETMQ-102-shutdown-await into 573b22c on apache:master.

coveralls · 2017-02-20T12:43:04Z

Coverage increased (+0.4%) to 31.919% when pulling 5667cdf on Jaskey:ROCKETMQ-102-shutdown-await into 573b22c on apache:master.

coveralls · 2017-02-20T12:43:04Z

Coverage increased (+0.4%) to 31.919% when pulling 5667cdf on Jaskey:ROCKETMQ-102-shutdown-await into 573b22c on apache:master.

coveralls · 2017-02-20T12:43:04Z

Coverage increased (+0.4%) to 31.919% when pulling 5667cdf on Jaskey:ROCKETMQ-102-shutdown-await into 573b22c on apache:master.

zhouxinyu · 2017-02-24T07:49:10Z

client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java

@@ -52,6 +56,7 @@
 import org.apache.rocketmq.common.protocol.header.PullMessageRequestHeader;
 import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.junit.After;
+import org.junit.Assert;


Hi, let's unify the assert tool and use org.assertj.core.api.Assertions.assertThat.

zhouxinyu · 2017-02-24T07:53:28Z

client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java

+        if (awaitTerminateMillis > 0) {
+            try {
+                this.consumeExecutor.awaitTermination(awaitTerminateMillis,TimeUnit.MILLISECONDS);
+                if (!this.consumeExecutor.isTerminated()) log.info("There are messages still being consumed in thread pool, but not going to await them anymore. Have awaited for {} ms",awaitTerminateMillis);


May be we need a common method to shutdown executor gracefully, like:

public static void shutdownGracefully(ExecutorService executor, long timeout, TimeUnit timeUnit) { executor.shutdown(); try { if(!executor.awaitTermination(timeout, timeUnit)) { executor.shutdownNow(); if(!executor.awaitTermination(timeout, timeUnit)) { LOG.warn(String.format("%s didn\'t terminate!", new Object[]{executor})); } } } catch (InterruptedException var5) { executor.shutdownNow(); Thread.currentThread().interrupt(); } }

We can, but where do we put this method in, in a Common Util?

add a new java file ThreadUtils?

Hi @Jaskey ,

I added a ThreadUtils, please refer to here.

Then maybe you merge that commit first and I will merge it later. Also, I don't think shutdownNow is a good choice for consume services, please refer to my comment

shutdown is the first choice, shutdownNow will be called if timeout.

And develop will be merged to master in next release, please refer to our new branching model.

@zhouxinyu In my option, rocketmq should have no right to interrupt what dev's business are doing, we may be doing some time-cost job which are doing transcation or inserting database, we should leave the task running if executor is still not terminated.

Besides, since the old version has not termination millis, so 0 of termination millis is the default behavior , shutdown now will cost task being interrupt/cancel immediately which is not proper in my opinion.

I have updated the pr , but still using my old method, please review and let's discuss more about it.

Jaskey · 2017-03-01T05:26:27Z

@zhouxinyu

Please check the updated pr.

Since shutdownNow needs developer to take respond to interrupts but actually they should not care about this and they will not do this in most of the cases since in the old version, no similar behavior will trigger, so I still remain using shutdown , and after timeout , the method will return and remains the very same flow.

BTW, unit test has been updated using AssetThat

coveralls · 2017-03-04T06:33:01Z

Changes Unknown when pulling 67cbdc3 on Jaskey:ROCKETMQ-102-shutdown-await into ** on apache:master**.

coveralls · 2017-03-04T06:33:01Z

Changes Unknown when pulling 67cbdc3 on Jaskey:ROCKETMQ-102-shutdown-await into ** on apache:master**.

coveralls · 2017-03-04T06:33:01Z

Changes Unknown when pulling 67cbdc3 on Jaskey:ROCKETMQ-102-shutdown-await into ** on apache:master**.

… consumed message, which may cause repeated messages. Add configuration to push consumer to accept await termination time to await consuming.

coveralls · 2017-03-07T06:53:05Z

Coverage increased (+0.3%) to 31.243% when pulling 3da9a79 on Jaskey:ROCKETMQ-102-shutdown-await into e3f4251 on apache:develop.

Jaskey · 2017-04-01T07:42:21Z

Repeated messages is a big problem for a message queue, and this is a known issue. When can this pr be reviewed and merged?
@lizhanhui @zhouxinyu @vongosling

vongosling · 2018-07-14T02:28:25Z

@Jaskey I will close the pr, if you happened to the same question, please let me know.

Jaskey · 2020-06-10T07:36:07Z

I request a another issue #2085 since I find this feature is still not supported for the latest release @zhouxinyu @vongosling

zhouxinyu reviewed Feb 17, 2017

View reviewed changes

zhouxinyu reviewed Feb 24, 2017

View reviewed changes

[ROCKETMQ-102] When shutdown(), the persisted offet is not the latest…

3da9a79

… consumed message, which may cause repeated messages. Add configuration to push consumer to accept await termination time to await consuming.

Jaskey changed the base branch from master to develop March 7, 2017 05:40

vongosling closed this Jul 14, 2018

[ROCKETMQ-102] When shutdown(), the persisted offet is not the latest consumed message, which may cause repeated messages. #64

[ROCKETMQ-102] When shutdown(), the persisted offet is not the latest consumed message, which may cause repeated messages. #64

Conversation

Jaskey commented Feb 16, 2017 • edited

lizhanhui commented Feb 16, 2017

Jaskey commented Feb 16, 2017 • edited

lizhanhui commented Feb 16, 2017

Jaskey commented Feb 16, 2017

Jaskey commented Feb 16, 2017 • edited

coveralls commented Feb 16, 2017 • edited

coveralls commented Feb 17, 2017

coveralls commented Feb 17, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhouxinyu commented Feb 17, 2017

coveralls commented Feb 17, 2017 • edited

coveralls commented Feb 17, 2017

Jaskey commented Feb 20, 2017

coveralls commented Feb 20, 2017

coveralls commented Feb 20, 2017

coveralls commented Feb 20, 2017

coveralls commented Feb 20, 2017

coveralls commented Feb 20, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jaskey Mar 7, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jaskey commented Mar 1, 2017 • edited

coveralls commented Mar 4, 2017

coveralls commented Mar 4, 2017

coveralls commented Mar 4, 2017

coveralls commented Mar 7, 2017 • edited

Jaskey commented Apr 1, 2017

vongosling commented Jul 14, 2018

Jaskey commented Jun 10, 2020 • edited

Jaskey commented Feb 16, 2017 •

edited

Jaskey commented Feb 16, 2017 •

edited

Jaskey commented Feb 16, 2017 •

edited

coveralls commented Feb 16, 2017 •

edited

coveralls commented Feb 17, 2017 •

edited

Jaskey Mar 7, 2017 •

edited

Jaskey commented Mar 1, 2017 •

edited

coveralls commented Mar 7, 2017 •

edited

Jaskey commented Jun 10, 2020 •

edited