Add Retry Mechanism to Handle Message Distribution Errors in NSQ

nsqd/http.go

@@ -245,9 +245,17 @@ func (s *httpServer) doPUB(w http.ResponseWriter, req *http.Request, ps httprout
 			return nil, http_api.Err{400, "INVALID_DEFER"}
 		}
 	}
+	var maxRetryChannel int
+	if mrc, ok := reqParams["max_retry_channel"]; ok {
+		maxRetryChannel, err = strconv.Atoi(mrc[0])
+		if err != nil {
+			return nil, http_api.Err{400, "INVALID_MAX_RETRY_CHANNEL"}
+		}
+	}
 
 	msg := NewMessage(topic.GenerateID(), body)
 	msg.deferred = deferred
+	msg.maxRetryChannel = maxRetryChannel
 	err = topic.PutMessage(msg)
 	if err != nil {
 		return nil, http_api.Err{503, "EXITING"}

nsqd/message.go

@@ -21,11 +21,12 @@ type Message struct {
 	Attempts  uint16
 
 	// for in-flight handling
-	deliveryTS time.Time
-	clientID   int64
-	pri        int64
-	index      int
-	deferred   time.Duration
+	deliveryTS      time.Time
+	clientID        int64
+	pri             int64
+	index           int
+	deferred        time.Duration
+	maxRetryChannel int
 }
 
 func NewMessage(id MessageID, body []byte) *Message {

nsqd/topic.go

@@ -330,11 +330,21 @@ func (t *Topic) messagePump() {
 				channel.PutMessageDeferred(chanMsg, chanMsg.deferred)
 				continue
 			}
-			err := channel.PutMessage(chanMsg)
-			if err != nil {
-				t.nsqd.logf(LOG_ERROR,
-					"TOPIC(%s) ERROR: failed to put msg(%s) to channel(%s) - %s",
-					t.name, msg.ID, channel.name, err)
+			checkRetry := 0
+			for {
+				err := channel.PutMessage(chanMsg)
+				if err != nil {
+					t.nsqd.logf(LOG_ERROR,
+						"TOPIC(%s) ERROR: failed to put msg(%s) to channel(%s) - %s",
+						t.name, msg.ID, channel.name, err)
+					if checkRetry >= msg.maxRetryChannel {
+						break
+					} else {
+						checkRetry++
+						continue
+					}
+				}
+				break
 			}
 		}
 	}