From 08ae522ff942cd9ff5386a73ead6b27846766a8f Mon Sep 17 00:00:00 2001
From: "zhaozhao.zz" <zhaozhao.zz@alibaba-inc.com>
Date: Thu, 11 Oct 2018 21:44:20 +0800
Subject: [PATCH] Streams: propagate lastid in XCLAIM when it has effect

---
 src/t_stream.c | 19 +++++++++++++------
 1 file changed, 13 insertions(+), 6 deletions(-)

diff --git a/src/t_stream.c b/src/t_stream.c
index 323783a7..16817c21 100644
--- a/src/t_stream.c
+++ b/src/t_stream.c
@@ -2075,6 +2075,8 @@ void xclaimCommand(client *c) {
     /* If we stopped because some IDs cannot be parsed, perhaps they
      * are trailing options. */
     time_t now = mstime();
+    streamID last_id = {0,0};
+    int lastid_updated = 0;
     for (; j < c->argc; j++) {
         int moreargs = (c->argc-1) - j; /* Number of additional arguments. */
         char *opt = c->argv[j]->ptr;
@@ -2100,18 +2102,18 @@ void xclaimCommand(client *c) {
                 != C_OK) return;
         } else if (!strcasecmp(opt,"LASTID") && moreargs) {
             j++;
-            streamID id;
-            if (streamParseStrictIDOrReply(c,c->argv[j],&id,0) != C_OK) return;
-            /* Technically it could be more correct to update that only after
-             * checking for syntax errors, but this option is only used by
-             * the replication command that outputs correct syntax. */
-            if (streamCompareID(&id,&group->last_id) > 0) group->last_id = id;
+            if (streamParseStrictIDOrReply(c,c->argv[j],&last_id,0) != C_OK) return;
         } else {
             addReplyErrorFormat(c,"Unrecognized XCLAIM option '%s'",opt);
             return;
         }
     }
 
+    if (streamCompareID(&last_id,&group->last_id) > 0) {
+        group->last_id = last_id;
+        lastid_updated = 1;
+    }
+
     if (deliverytime != -1) {
         /* If a delivery time was passed, either with IDLE or TIME, we
          * do some sanity check on it, and set the deliverytime to now
@@ -2132,6 +2134,7 @@ void xclaimCommand(client *c) {
     streamConsumer *consumer = streamLookupConsumer(group,c->argv[3]->ptr,1);
     void *arraylenptr = addDeferredMultiBulkLength(c);
     size_t arraylen = 0;
+    long long dirty = server.dirty;
     for (int j = 5; j <= last_id_arg; j++) {
         streamID id;
         unsigned char buf[sizeof(streamID)];
@@ -2198,6 +2201,10 @@ void xclaimCommand(client *c) {
             server.dirty++;
         }
     }
+    if (server.dirty == dirty && lastid_updated) {
+        streamPropagateGroupID(c,c->argv[1],group,c->argv[2]);
+        server.dirty++;
+    }
     setDeferredMultiBulkLength(c,arraylenptr,arraylen);
     preventCommandPropagation(c);
 }