Add IsTimeout method to Error (#903)

This convenience method just checks if Error code is
ErrTimedOut/ErrTimedOutQueue.

Also modifies the example in README.md.

Modifies the CHANGELOG.md.

CHANGELOG.md

@@ -9,6 +9,8 @@ This is a maintenance release:
    will be used for subsequent (new) connections to a broker.
  * Channel based producer (Producer `ProduceChannel()`) and channel based
    consumer (Consumer `Events()`) are deprecated.
+ * Added `IsTimeout()` on Error type. This is a convenience method that checks
+   if the error is due to a timeout.
 
 
 ## v1.9.2

README.md

@@ -68,10 +68,10 @@ func main() {
 		msg, err := c.ReadMessage(time.Second)
 		if err == nil {
 			fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))
-		} else if err.(kafka.Error).Code() != kafka.ErrTimedOut {
+		} else if !err.(kafka.Error).IsTimeout() {
 			// The client will automatically try to recover from all errors.
-			// kafka.ErrTimedOut is not considered an error because it is
-			// raised by ReadMessage on timeout.
+			// Timeout is not considered an error because it is raised by
+			// ReadMessage in absence of messages.
 			fmt.Printf("Consumer error: %v (%v)\n", err, msg)
 		}
 	}

kafka/consumer.go

@@ -367,7 +367,7 @@ func (c *Consumer) Logs() chan LogEvent {
 // a new message or error. `timeout` may be set to -1 for
 // indefinite wait.
 //
-// Timeout is returned as (nil, err) where err is `err.(kafka.Error).Code() == kafka.ErrTimedOut`.
+// Timeout is returned as (nil, err) where `err.(kafka.Error).IsTimeout() == true`.
 //
 // Messages are returned as (msg, nil),
 // while general errors are returned as (nil, err),

kafka/consumer_test.go

@@ -204,6 +204,21 @@ func TestConsumerAPIs(t *testing.T) {
 		t.Errorf("Committed() failed but returned non-nil Offsets: %s\n", offsets)
 	}
 
+	// Test timeouts using ReadMessage.
+	msg, err := c.ReadMessage(time.Millisecond)
+	t.Logf("ReadMessage() returned message %s and error %s\n", msg, err)
+
+	// Check both ErrTimedOut and IsTimeout() to ensure they're consistent.
+	if err == nil || !err.(Error).IsTimeout() {
+		t.Errorf("ReadMessage() should time out, instead got %s\n", err)
+	}
+	if err == nil || err.(Error).Code() != ErrTimedOut {
+		t.Errorf("ReadMessage() should time out, instead got %s\n", err)
+	}
+	if msg != nil {
+		t.Errorf("ReadMessage() should not return a message in case of error\n")
+	}
+
 	err = c.Close()
 	if err != nil {
 		t.Errorf("Close failed: %s", err)
@@ -295,6 +310,9 @@ func TestConsumerAssignment(t *testing.T) {
 		if err.(Error).Code() != ErrTimedOut {
 			t.Errorf("Expected ReadMessage to fail with ErrTimedOut, not %v", err)
 		}
+		if !err.(Error).IsTimeout() {
+			t.Errorf("Expected ReadMessage to fail with a timeout error, not %v", err)
+		}
 
 		if tmout == 0 {
 			if duration.Seconds() > 0.1 {

kafka/error.go

@@ -127,6 +127,12 @@ func (e Error) IsRetriable() bool {
 	return e.retriable
 }
 
+// IsTimeout returns true if the error is a timeout error.
+// A timeout error indicates that the operation timed out locally.
+func (e Error) IsTimeout() bool {
+	return e.code == ErrTimedOut || e.code == ErrTimedOutQueue
+}
+
 // TxnRequiresAbort returns true if the error is an abortable transaction error
 // that requires the application to abort the current transaction with
 // AbortTransaction() and start a new transaction with BeginTransaction()