Merge pull request #1015 from cserwen/orderly_consume_github

[ISSUE #994] Fix: unlock all queues when consumer shutdown in orderly model
apache · Mar 20, 2023 · 3b8b238 · 3b8b238
2 parents 56d1a7d + 9bca88d
commit 3b8b238
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 7 deletions.
diff --git a/consumer/consumer.go b/consumer/consumer.go
@@ -631,12 +631,12 @@ func (dc *defaultConsumer) doUnlock(addr string, body *lockBatchRequestBody, one
 		}
 	} else {
 		response, err := dc.client.InvokeSync(context.Background(), addr, request, 1*time.Second)
-		rlog.Error("lock MessageQueue to broker invoke error", map[string]interface{}{
-			rlog.LogKeyBroker:        addr,
-			rlog.LogKeyUnderlayError: err,
-		})
-		if response.Code != internal.ResSuccess {
-			// TODO error
+		if err != nil || response == nil || response.Code != internal.ResSuccess {
+			rlog.Error("lock MessageQueue to broker invoke error", map[string]interface{}{
+				rlog.LogKeyBroker:        addr,
+				rlog.LogKeyUnderlayError: err,
+				"response":               response,
+			})
 		}
 	}
 }

diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
@@ -277,7 +277,9 @@ func (pc *pushConsumer) Shutdown() error {
 			pc.option.TraceDispatcher.Close()
 		}
 		close(pc.done)
-
+		if pc.consumeOrderly && pc.model == Clustering {
+			pc.unlockAll(false)
+		}
 		pc.client.UnregisterConsumer(pc.consumerGroup)
 		err = pc.defaultConsumer.shutdown()
 	})