Revert "Nettest fix (#788)" (#824)

This reverts commit 2e02f37.

Author: rhansen2

.circleci/config.yml

@@ -184,6 +184,17 @@ jobs:
     working_directory: *working_directory
     environment:
       KAFKA_VERSION: "2.4.1"
+
+      # Need to skip nettest to avoid these kinds of errors:
+      #  --- FAIL: TestConn/nettest (17.56s)
+      #    --- FAIL: TestConn/nettest/PingPong (7.40s)
+      #      conntest.go:112: unexpected Read error: [7] Request Timed Out: the request exceeded the user-specified time limit in the request
+      #      conntest.go:118: mismatching value: got 77, want 78
+      #      conntest.go:118: mismatching value: got 78, want 79
+      # ...
+      #
+      # TODO: Figure out why these are happening and fix them (they don't appear to be new).
+      KAFKA_SKIP_NETTEST: "1"
     docker:
     - image: circleci/golang
     - image: wurstmeister/zookeeper
@@ -200,6 +211,17 @@ jobs:
     working_directory: *working_directory
     environment:
       KAFKA_VERSION: "2.6.0"
+
+      # Need to skip nettest to avoid these kinds of errors:
+      #  --- FAIL: TestConn/nettest (17.56s)
+      #    --- FAIL: TestConn/nettest/PingPong (7.40s)
+      #      conntest.go:112: unexpected Read error: [7] Request Timed Out: the request exceeded the user-specified time limit in the request
+      #      conntest.go:118: mismatching value: got 77, want 78
+      #      conntest.go:118: mismatching value: got 78, want 79
+      # ...
+      #
+      # TODO: Figure out why these are happening and fix them (they don't appear to be new).
+      KAFKA_SKIP_NETTEST: "1"
     docker:
     - image: circleci/golang
     - image: wurstmeister/zookeeper
@@ -216,6 +238,17 @@ jobs:
     working_directory: *working_directory
     environment:
       KAFKA_VERSION: "2.7.1"
+
+      # Need to skip nettest to avoid these kinds of errors:
+      #  --- FAIL: TestConn/nettest (17.56s)
+      #    --- FAIL: TestConn/nettest/PingPong (7.40s)
+      #      conntest.go:112: unexpected Read error: [7] Request Timed Out: the request exceeded the user-specified time limit in the request
+      #      conntest.go:118: mismatching value: got 77, want 78
+      #      conntest.go:118: mismatching value: got 78, want 79
+      # ...
+      #
+      # TODO: Figure out why these are happening and fix them (they don't appear to be new).
+      KAFKA_SKIP_NETTEST: "1"
     docker:
     - image: circleci/golang
     - image: wurstmeister/zookeeper

conn.go

@@ -133,9 +133,11 @@ const (
 	ReadCommitted   IsolationLevel = 1
 )
 
-// DefaultClientID is the default value used as ClientID of kafka
-// connections.
-var DefaultClientID string
+var (
+	// DefaultClientID is the default value used as ClientID of kafka
+	// connections.
+	DefaultClientID string
+)
 
 func init() {
 	progname := filepath.Base(os.Args[0])
@@ -261,12 +263,10 @@ func (c *Conn) Controller() (broker Broker, err error) {
 			}
 			for _, brokerMeta := range res.Brokers {
 				if brokerMeta.NodeID == res.ControllerID {
-					broker = Broker{
-						ID:   int(brokerMeta.NodeID),
+					broker = Broker{ID: int(brokerMeta.NodeID),
 						Port: int(brokerMeta.Port),
 						Host: brokerMeta.Host,
-						Rack: brokerMeta.Rack,
-					}
+						Rack: brokerMeta.Rack}
 					break
 				}
 			}
@@ -322,6 +322,7 @@ func (c *Conn) findCoordinator(request findCoordinatorRequestV0) (findCoordinato
 	err := c.readOperation(
 		func(deadline time.Time, id int32) error {
 			return c.writeRequest(findCoordinator, v0, id, request)
+
 		},
 		func(deadline time.Time, size int) error {
 			return expectZeroSize(func() (remain int, err error) {
@@ -751,8 +752,9 @@ func (c *Conn) ReadBatch(minBytes, maxBytes int) *Batch {
 // ReadBatchWith in every way is similar to ReadBatch. ReadBatch is configured
 // with the default values in ReadBatchConfig except for minBytes and maxBytes.
 func (c *Conn) ReadBatchWith(cfg ReadBatchConfig) *Batch {
+
 	var adjustedDeadline time.Time
-	maxFetch := int(c.fetchMaxBytes)
+	var maxFetch = int(c.fetchMaxBytes)
 
 	if cfg.MinBytes < 0 || cfg.MinBytes > maxFetch {
 		return &Batch{err: fmt.Errorf("kafka.(*Conn).ReadBatch: minBytes of %d out of [1,%d] bounds", cfg.MinBytes, maxFetch)}
@@ -857,7 +859,11 @@ func (c *Conn) ReadBatchWith(cfg ReadBatchConfig) *Batch {
 
 	var msgs *messageSetReader
 	if err == nil {
-		msgs, err = newMessageSetReader(&c.rbuf, remain)
+		if highWaterMark == offset {
+			msgs = &messageSetReader{empty: true}
+		} else {
+			msgs, err = newMessageSetReader(&c.rbuf, remain)
+		}
 	}
 	if err == errShortRead {
 		err = checkTimeoutErr(adjustedDeadline)
@@ -953,6 +959,7 @@ func (c *Conn) readOffset(t int64) (offset int64, err error) {
 // connection. If there are none, the method fetches all partitions of the kafka
 // cluster.
 func (c *Conn) ReadPartitions(topics ...string) (partitions []Partition, err error) {
+
 	if len(topics) == 0 {
 		if len(c.topic) != 0 {
 			defaultTopics := [...]string{c.topic}
@@ -1181,6 +1188,7 @@ func (c *Conn) writeCompressedMessages(codec CompressionCodec, msgs ...Message)
 						}
 						return size, err
 					}
+
 				})
 				if err != nil {
 					return size, err
@@ -1548,7 +1556,7 @@ func (c *Conn) saslAuthenticate(data []byte) ([]byte, error) {
 		return nil, err
 	}
 	if version == v1 {
-		request := saslAuthenticateRequestV0{Data: data}
+		var request = saslAuthenticateRequestV0{Data: data}
 		var response saslAuthenticateResponseV0
 
 		err := c.writeOperation(