Raise error when the reader cannot consume anymore

.golanci.yaml

@@ -15,4 +15,4 @@ linters:
     - exhaustruct
     - gocognit
     - gochecknoinits
-    - gocyclo
+    - gocyclo

go.mod

@@ -65,7 +65,7 @@ require (
 	go.opentelemetry.io/otel/trace v1.24.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.1.0 // indirect
 	golang.org/x/crypto v0.21.0 // indirect
-	golang.org/x/net v0.22.0 // indirect
+	golang.org/x/net v0.23.0 // indirect
 	golang.org/x/sync v0.6.0 // indirect
 	golang.org/x/sys v0.18.0 // indirect
 	golang.org/x/text v0.14.0 // indirect

reader.go

@@ -336,15 +336,15 @@ func (k *Kafka) consume(
 
 			err = NewXk6KafkaError(noMoreMessages, "No more messages.", nil)
 			logger.WithField("error", err).Info(err)
-			return messages
+			common.Throw(k.vu.Runtime(), err)
 		}
 
 		if err != nil {
 			k.reportReaderStats(reader.Stats())
 
 			err = NewXk6KafkaError(failedReadMessage, "Unable to read messages.", nil)
 			logger.WithField("error", err).Error(err)
-			return messages
+			common.Throw(k.vu.Runtime(), err)
 		}
 
 		var messageTime string

reader_test.go

@@ -33,7 +33,7 @@ func TestConsumerMaxWaitExceeded(t *testing.T) {
 		require.NoError(t, test.moveToVUCode())
 
 		// Consume a message in the VU function.
-		assert.NotPanics(t, func() {
+		assert.Panics(t, func() {
 			messages := test.module.Kafka.consume(reader, &ConsumeConfig{Limit: 1})
 			assert.Empty(t, messages)
 		})
@@ -231,7 +231,7 @@ func TestConsumerContextCancelled(t *testing.T) {
 		test.cancelContext()
 
 		// Consume a message in the VU function.
-		assert.NotPanics(t, func() {
+		assert.Panics(t, func() {
 			messages := test.module.Kafka.consume(reader, &ConsumeConfig{Limit: 1})
 			assert.Empty(t, messages)
 		})

scripts/test_json.js

@@ -114,36 +114,40 @@ export default function () {
     writer.produce({ messages: messages });
   }
 
-  // Read 10 messages only
-  let messages = reader.consume({ limit: 10 });
+  try {
+    // Read 10 messages only
+    let messages = reader.consume({ limit: 10 });
 
-  check(messages, {
-    "10 messages are received": (messages) => messages.length == 10,
-  });
+    check(messages, {
+      "10 messages are received": (messages) => messages.length == 10,
+    });
 
-  check(messages[0], {
-    "Topic equals to xk6_kafka_json_topic": (msg) => msg["topic"] == topic,
-    "Key contains key/value and is JSON": (msg) =>
-      schemaRegistry
-        .deserialize({ data: msg.key, schemaType: SCHEMA_TYPE_JSON })
-        .correlationId.startsWith("test-id-"),
-    "Value contains key/value and is JSON": (msg) =>
-      typeof schemaRegistry.deserialize({
-        data: msg.value,
-        schemaType: SCHEMA_TYPE_JSON,
-      }) == "object" &&
-      schemaRegistry.deserialize({
-        data: msg.value,
-        schemaType: SCHEMA_TYPE_JSON,
-      }).name == "xk6-kafka",
-    "Header equals {'mykey': 'myvalue'}": (msg) =>
-      "mykey" in msg.headers &&
-      String.fromCharCode(...msg.headers["mykey"]) == "myvalue",
-    "Time is past": (msg) => new Date(msg["time"]) < new Date(),
-    "Partition is zero": (msg) => msg["partition"] == 0,
-    "Offset is gte zero": (msg) => msg["offset"] >= 0,
-    "High watermark is gte zero": (msg) => msg["highWaterMark"] >= 0,
-  });
+    check(messages[0], {
+      "Topic equals to xk6_kafka_json_topic": (msg) => msg["topic"] == topic,
+      "Key contains key/value and is JSON": (msg) =>
+        schemaRegistry
+          .deserialize({ data: msg.key, schemaType: SCHEMA_TYPE_JSON })
+          .correlationId.startsWith("test-id-"),
+      "Value contains key/value and is JSON": (msg) =>
+        typeof schemaRegistry.deserialize({
+          data: msg.value,
+          schemaType: SCHEMA_TYPE_JSON,
+        }) == "object" &&
+        schemaRegistry.deserialize({
+          data: msg.value,
+          schemaType: SCHEMA_TYPE_JSON,
+        }).name == "xk6-kafka",
+      "Header equals {'mykey': 'myvalue'}": (msg) =>
+        "mykey" in msg.headers &&
+        String.fromCharCode(...msg.headers["mykey"]) == "myvalue",
+      "Time is past": (msg) => new Date(msg["time"]) < new Date(),
+      "Partition is zero": (msg) => msg["partition"] == 0,
+      "Offset is gte zero": (msg) => msg["offset"] >= 0,
+      "High watermark is gte zero": (msg) => msg["highWaterMark"] >= 0,
+    });
+  } catch (error) {
+    console.error(error);
+  }
 }
 
 export function teardown(data) {