FrancoisPoinsot před 6 roky
rodič
revize
eb3cb9538e
1 změnil soubory, kde provedl 6 přidání a 1 odebrání
  1. 6 1
      consumer.go

+ 6 - 1
consumer.go

@@ -638,7 +638,12 @@ func (child *partitionConsumer) parseResponse(response *FetchResponse) ([]*Consu
 			// control record
 			isControl, err := records.isControl()
 			if err != nil {
-				//TODO maybe we should handle this ? a log at least
+				// I don't know why there is this continue in case of error to begin with
+				// Safe bet is to ignore control messages if ReadUncommitted
+				// and block on them in case of error and ReadCommitted
+				if child.conf.Consumer.IsolationLevel == ReadCommitted {
+					return nil, err
+				}
 				continue
 			}
 			if isControl {