Fix NotLeaderForPartition error (segmentio#19)

* Fix NotLeaderForPartition error
huangfu · Sep 26, 2017 · c6ecc9b · c6ecc9b
1 parent 56ac8b3
commit c6ecc9b
Showing 1 changed file with 9 additions and 0 deletions.
diff --git a/reader.go b/reader.go
@@ -424,7 +424,16 @@ func (r *reader) run(ctx context.Context, offset int64, join *sync.WaitGroup) {
 			switch offset, err = r.read(ctx, offset, conn); err {
 			case nil:
 				errcount = 0
+			case NotLeaderForPartition:
+				r.withLogger(func(log *log.Logger) {
+					log.Printf("failed to read from current broker for partition %d of %s at offset %d, not the leader", r.partition, r.topic, offset)
+				})
+
+				conn.Close()
 
+				// The next call to .initialize will re-establish a connection to the proper
+				// partition leader.
+				break readLoop
 			case RequestTimedOut:
 				// Timeout on the kafka side, this can be safely retried.
 				errcount = 0