Some changes in constants.

Author: nachogiljaldo

conn.go

@@ -17,6 +17,8 @@ import (
 var (
 	errInvalidWriteTopic     = errors.New("writes must NOT set Topic on kafka.Message")
 	errInvalidWritePartition = errors.New("writes must NOT set Partition on kafka.Message")
+
+	undefinedGenerationId int32 = -1
 )
 
 // Conn represents a connection to a kafka broker.
@@ -184,7 +186,7 @@ func NewConnWith(conn net.Conn, config ConnConfig) *Conn {
 		offset:          FirstOffset,
 		requiredAcks:    -1,
 		transactionalID: emptyToNullable(config.TransactionalID),
-		generationId:    -1,
+		generationId:    undefinedGenerationId,
 	}
 
 	c.wb.w = &c.wbuf

reader.go

@@ -156,7 +156,7 @@ func (r *Reader) commitOffsetsWithRetry(gen *Generation, offsetStash offsetStash
 		for partition, commitInfo := range partitionsInfo {
 			// if there is a generation and it is different than the current
 			// it means there was a rebalance
-			if commitInfo.generationId != -1 {
+			if commitInfo.generationId != undefinedGenerationId {
 				if gen.ID == commitInfo.generationId {
 					msgsForTopic[partition] = commitInfo.offset
 				} else {
@@ -855,7 +855,7 @@ func (r *Reader) FetchMessage(ctx context.Context) (Message, error) {
 		r.mutex.Lock()
 
 		if !r.closed && r.version == 0 {
-			r.start(-1, r.getTopicPartitionOffset())
+			r.start(undefinedGenerationId, r.getTopicPartitionOffset())
 		}
 
 		version := r.version
@@ -1076,7 +1076,7 @@ func (r *Reader) SetOffset(offset int64) error {
 		r.offset = offset
 
 		if r.version != 0 {
-			r.start(-1, r.getTopicPartitionOffset())
+			r.start(undefinedGenerationId, r.getTopicPartitionOffset())
 		}
 
 		r.activateReadLag()