Explorar o código

consumer: don't check LastStableOffset

It's only needed for transactions (which we don't support yet) and the
logic was wrong anyway. Fixes slow consuming for certain Kafka 0.11
configurations.
Evan Huus %!s(int64=8) %!d(string=hai) anos
pai
achega
10f33bec2a
Modificáronse 1 ficheiros con 0 adicións e 5 borrados
  1. 0 5
      consumer.go

+ 0 - 5
consumer.go

@@ -546,11 +546,6 @@ func (child *partitionConsumer) parseRecords(block *FetchResponseBlock) ([]*Cons
 		} else {
 		} else {
 			incomplete = true
 			incomplete = true
 		}
 		}
-
-		if child.offset > block.LastStableOffset {
-			// We reached the end of closed transactions
-			break
-		}
 	}
 	}
 
 
 	if incomplete || len(messages) == 0 {
 	if incomplete || len(messages) == 0 {