Fix benchmarks

joel-rieke · Jul 12, 2018 · 7faaf2b · 7faaf2b
1 parent 536dc15
commit 7faaf2b
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 6 deletions.
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -1,16 +1,16 @@
 kafka:
-  image: wurstmeister/kafka
+  image: wurstmeister/kafka:1.1.0
   links:
     - zookeeper
   ports:
     - "9092:9092"
   environment:
-    KAFKA_VERSION: "0.10.1.0"
     KAFKA_CREATE_TOPICS: "test-writer-0:3:1,test-writer-1:3:1"
     KAFKA_ADVERTISED_HOST_NAME: "localhost"
     KAFKA_ADVERTISED_PORT: "9092"
     KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
     KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
+    KAFKA_MESSAGE_MAX_BYTES: 200000000
 
 zookeeper:
   image: wurstmeister/zookeeper

diff --git a/read.go b/read.go
@@ -361,6 +361,7 @@ func readMessageHeader(r *bufio.Reader, sz int) (offset int64, attributes int8,
 	switch version {
 	case 0:
 	case 1:
+	case 2:
 		remain, err = readInt64(r, remain, &timestamp)
 	default:
 		err = fmt.Errorf("unsupported message version %d found in fetch response", version)

diff --git a/reader_test.go b/reader_test.go
@@ -385,7 +385,7 @@ func prepareReader(t *testing.T, ctx context.Context, r *Reader, msgs ...Message
 var (
 	benchmarkReaderOnce    sync.Once
 	benchmarkReaderTopic   = makeTopic()
-	benchmarkReaderPayload = make([]byte, 16*1024)
+	benchmarkReaderPayload = make([]byte, 2*1024)
 )
 
 func BenchmarkReader(b *testing.B) {
@@ -422,11 +422,14 @@ func BenchmarkReader(b *testing.B) {
 		MaxWait:   100 * time.Millisecond,
 	})
 
-	for i := 0; i != b.N; i++ {
+	for i := 0; i < b.N; i++ {
 		if (i % 10000) == 0 {
-			r.SetOffset(0)
+			r.SetOffset(-1)
+		}
+		_, err := r.ReadMessage(ctx)
+		if err != nil {
+			b.Fatal(err)
 		}
-		r.ReadMessage(ctx)
 	}
 
 	r.Close()