Fix for Dialer.LookupPartition (#191)

* Fix lookup partition (#1) * Test against multiple Kafka versions * Message writer for record batches (#163) * Add branching for record batch writer * Make key and value from headers public * Add writeVarInt function * Implement writeRecord method * Write record batch method * Use zigzag encoding for varint * Drop logging * Add unit test for v2 record batch * Remove commented out code * Use int constants from math package * Use exact record size instead of extra buffer * Give a simpler name to calcVarIntLen * Move error check out of if else block * Extract message set size in a function * Give calcRecordSize a shorter name * Protect recordSize from confusing order of params * Use recordSize intead of estimatedRecordSize * Extract record batch size * Use record size from calculation * Assign offsets to messages in record batch * Write message set straight to output writer * Write record batch straight to output writer * Get rid of remainder buffer * Remove commented out code * Use bytes.Equal for comparing slices of bytes * Compress messages before calculating size * Run tests for kafka 0.10 (#187) Added a utility method to check broker version so that tests for newer functionality can be skipped. * fix * Fix memory leak in LookupPartition * Fix ReadPartitions
segmentio · Feb 11, 2019 · a3bd32e · a3bd32e
1 parent e956173
commit a3bd32e
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 2 deletions.
diff --git a/conn.go b/conn.go
@@ -755,7 +755,7 @@ func (c *Conn) ReadPartitions(topics ...string) (partitions []Partition, err err
 			}
 
 			for _, t := range res.Topics {
-				if t.TopicErrorCode != 0 && t.TopicName == c.topic {
+				if t.TopicErrorCode != 0 && (c.topic == "" || t.TopicName == c.topic) {
 					// We only report errors if they happened for the topic of
 					// the connection, otherwise the topic will simply have no
 					// partitions in the result set.

diff --git a/dialer.go b/dialer.go
@@ -154,7 +154,10 @@ func (d *Dialer) LookupPartition(ctx context.Context, network string, address st
 	go func() {
 		for attempt := 0; true; attempt++ {
 			if attempt != 0 {
-				sleep(ctx, backoff(attempt, 100*time.Millisecond, 10*time.Second))
+				if !sleep(ctx, backoff(attempt, 100*time.Millisecond, 10*time.Second)) {
+					errch <- ctx.Err()
+					return
+				}
 			}
 
 			partitions, err := c.ReadPartitions(topic)