Consumer timestamp test adjusted

consumer.go

@@ -795,6 +795,9 @@ func (bc *brokerConsumer) fetchNewMessages() (*FetchResponse, error) {
 		MinBytes:    bc.consumer.conf.Consumer.Fetch.Min,
 		MaxWaitTime: int32(bc.consumer.conf.Consumer.MaxWaitTime / time.Millisecond),
 	}
+	if bc.consumer.conf.Version.IsAtLeast(V0_9_0_0) {
+		request.Version = 1
+	}
 	if bc.consumer.conf.Version.IsAtLeast(V0_10_0_0) {
 		request.Version = 2
 	}

consumer_test.go

@@ -999,7 +999,15 @@ func TestConsumerTimestamps(t *testing.T) {
 		messages          []testMessage
 		expectedTimestamp []time.Time
 	}{
-		{V0_8_2_0, false, []testMessage{
+		{MinVersion, false, []testMessage{
+			{nil, testMsg, 1, now},
+			{nil, testMsg, 2, now},
+		}, []time.Time{{}, {}}},
+		{V0_9_0_0, false, []testMessage{
+			{nil, testMsg, 1, now},
+			{nil, testMsg, 2, now},
+		}, []time.Time{{}, {}}},
+		{V0_10_0_0, false, []testMessage{
 			{nil, testMsg, 1, now},
 			{nil, testMsg, 2, now},
 		}, []time.Time{{}, {}}},
@@ -1024,26 +1032,33 @@ func TestConsumerTimestamps(t *testing.T) {
 		var offsetResponseVersion int16
 		cfg := NewConfig()
 		cfg.Version = d.kversion
-		switch d.kversion {
-		default:
-			fr = &FetchResponse{}
+		switch {
+		case d.kversion.IsAtLeast(V0_11_0_0):
+			offsetResponseVersion = 1
+			fr = &FetchResponse{Version: 4, LogAppendTime: d.logAppendTime, Timestamp: now}
 			for _, m := range d.messages {
-				fr.AddMessageWithTimestamp("my_topic", 0, m.key, testMsg, m.offset, m.timestamp, 0)
+				fr.AddRecordWithTimestamp("my_topic", 0, m.key, testMsg, m.offset, m.timestamp)
 			}
-		case V0_10_2_1:
+			fr.SetLastOffsetDelta("my_topic", 0, 2)
+			fr.SetLastStableOffset("my_topic", 0, 2)
+		case d.kversion.IsAtLeast(V0_10_1_0):
 			offsetResponseVersion = 1
 			fr = &FetchResponse{Version: 3, LogAppendTime: d.logAppendTime, Timestamp: now}
 			for _, m := range d.messages {
 				fr.AddMessageWithTimestamp("my_topic", 0, m.key, testMsg, m.offset, m.timestamp, 1)
 			}
-		case V0_11_0_0:
-			offsetResponseVersion = 1
-			fr = &FetchResponse{Version: 4, LogAppendTime: d.logAppendTime, Timestamp: now}
+		default:
+			var version int16
+			switch {
+			case d.kversion.IsAtLeast(V0_10_0_0):
+				version = 2
+			case d.kversion.IsAtLeast(V0_9_0_0):
+				version = 1
+			}
+			fr = &FetchResponse{Version: version}
 			for _, m := range d.messages {
-				fr.AddRecordWithTimestamp("my_topic", 0, m.key, testMsg, m.offset, m.timestamp)
+				fr.AddMessageWithTimestamp("my_topic", 0, m.key, testMsg, m.offset, m.timestamp, 0)
 			}
-			fr.SetLastOffsetDelta("my_topic", 0, 2)
-			fr.SetLastStableOffset("my_topic", 0, 2)
 		}
 
 		broker0 := NewMockBroker(t, 0)

message.go

@@ -5,12 +5,15 @@ import (
 	"time"
 )
 
-// CompressionCodec represents the various compression codecs recognized by Kafka in messages.
-type CompressionCodec int8
-
 // The lowest 3 bits contain the compression codec used for the message
 const compressionCodecMask int8 = 0x07
 
+// Bit 3 set for "LogAppend" timestamps
+const timestampTypeMask = 0x08
+
+// CompressionCodec represents the various compression codecs recognized by Kafka in messages.
+type CompressionCodec int8
+
 const (
 	CompressionNone   CompressionCodec = 0
 	CompressionGZIP   CompressionCodec = 1
@@ -19,8 +22,6 @@ const (
 	CompressionZSTD   CompressionCodec = 4
 )
 
-const timestampTypeMask = 0x08
-
 func (cc CompressionCodec) String() string {
 	return []string{
 		"none",