kafka消费异常抛出

Author: team-ide

kafka/kafka.go

@@ -217,29 +217,28 @@ func (this_ *Service) Pull(groupId string, topics []string, PullSize int, PullTi
 	if err != nil {
 		return
 	}
+	defer func() {
+		e := group.Close()
+		if e != nil {
+			util.Logger.Error("group close error", zap.Error(e))
+		}
+	}()
 	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(PullTimeout))
 	handler := &consumerGroupHandler{
 		size:   PullSize,
 		cancel: cancel,
 	}
 	util.Logger.Info("kafka pull start", zap.Any("topics", topics), zap.Any("groupId", groupId), zap.Any("timeout", PullTimeout))
 	err = group.Consume(ctx, topics, handler)
-
 	util.Logger.Info("kafka pull end", zap.Any("topics", topics), zap.Any("groupId", groupId), zap.Any("timeout", PullTimeout))
 	if err != nil {
 		util.Logger.Error("group consume error", zap.Error(err))
+		return
 	}
 
-	err = group.Close()
-	if err != nil {
-		util.Logger.Error("group close error", zap.Error(err))
-	}
 	for _, one := range handler.messages {
 		var msg *Message
-		msg, err = ConsumerMessageToMessage(keyType, valueType, one)
-		if err != nil {
-			return
-		}
+		msg = ConsumerMessageToMessage(keyType, valueType, one)
 		msgList = append(msgList, msg)
 	}
 	return
@@ -901,7 +900,7 @@ func MessageToProducerMessage(msg *Message) (producerMessage *sarama.ProducerMes
 	return
 }
 
-func ConsumerMessageToMessage(keyType string, valueType string, consumerMessage *sarama.ConsumerMessage) (msg *Message, err error) {
+func ConsumerMessageToMessage(keyType string, valueType string, consumerMessage *sarama.ConsumerMessage) (msg *Message) {
 	var key string
 	var value string
 

package.json

@@ -1,3 +1,3 @@
 {
-  "version": "1.2.41"
+  "version": "1.2.42"
 }