Fix offsets

Dockerfile

@@ -4,7 +4,7 @@ ARG VERSION_TAG
 ARG TARGETOS
 ARG TARGETARCH
 
-RUN apk add --no-cache ca-certificates=20240705-r0 openssl=3.3.2-r3 && \
+RUN apk add --no-cache ca-certificates=20240705-r0 openssl=3.3.2-r1 && \
     adduser -D -u 12345 -g 12345 k6
 COPY ./dist/xk6-kafka_${VERSION_TAG}_${TARGETOS}_${TARGETARCH} /usr/bin/k6
 

README.md

@@ -461,6 +461,45 @@ The example scripts are available as `test_<format/feature>.js` with more code a
 13. What is the difference between hard-coded schemas in the script and the ones fetched from the Schema Registry?
     Read [this comment](https://github.com/mostafa/xk6-kafka/issues/298#issuecomment-2165246467).
 
+14. I want to specify the offset of a message when consuming from a topic. How can I do that?
+    To specify the offset of a message while consuming from a topic, use the following options based on your consumption setup:
+
+    1. **When consuming from a group:**
+    Use the `startOffset` option in the `Reader` object. This option allows you to define the starting point for message consumption. Here are the values you can use for `startOffset`:
+
+       - `-1`: Consume from the most recent message. This is equivalent to `START_OFFSETS_LAST_OFFSET`.
+       - `-2`: Consume from the oldest message. This is equivalent to `START_OFFSETS_FIRST_OFFSET`.
+       - Any positive number: Consume from the specific offset number provided.
+
+        The constants `START_OFFSETS_LAST_OFFSET` and `START_OFFSETS_FIRST_OFFSET` are part of the xk6-kafka module. You can import and use them in your script. The `startOffset` option is a string.
+
+        ```javascript
+        import {
+            Reader,
+            START_OFFSETS_LAST_OFFSET,
+        } from "k6/x/kafka";
+
+        const reader = new Reader({
+            brokers: ["localhost:9092"], // Replace with your broker(s)
+            groupID: "example-group", // Specify your consumer group ID
+            groupTopics: ["example-topic"], // List of topics for the group
+            startOffset: START_OFFSETS_LAST_OFFSET, // Use the most recent offset
+        });
+        ```
+
+    2. **When consuming from a topic:**
+    Use the `offset` option instead of `startOffset`. The `offset` option is a number that directly specifies the offset of the message you want to consume, unlike `startOffset`, which is a string.
+
+        ```javascript
+        import { Reader } from "k6/x/kafka";
+
+        const reader = new Reader({
+            brokers: ["localhost:9092"], // Replace with your broker(s)
+            topic: "example-topic", // Specify the topic
+            offset: 10, // Consume from offset 10
+        });
+        ```
+
 ## Contributions, Issues and Feedback
 
 I'd be thrilled to receive contributions and feedback on this project. You're always welcome to create an issue if you find one (or many). I would do my best to address the issues. Also, feel free to contribute by opening a PR with changes, and I'll do my best to review and merge it as soon as I can.

error_codes.go

@@ -31,11 +31,12 @@ const (
 	failedDecodeBase64         errCode = 2009
 
 	// consumer.
-	failedSetOffset     errCode = 3000
-	failedReadMessage   errCode = 3001
-	noMoreMessages      errCode = 3002
-	partitionAndGroupID errCode = 3003
-	topicAndGroupID     errCode = 3004
+	failedSetOffset        errCode = 3000
+	failedReadMessage      errCode = 3001
+	noMoreMessages         errCode = 3002
+	partitionAndGroupID    errCode = 3003
+	topicAndGroupID        errCode = 3004
+	failedParseStartOffset errCode = 3005
 
 	// authentication.
 	failedCreateDialerWithScram   errCode = 4000

go.mod

@@ -5,32 +5,32 @@ go 1.23
 toolchain go1.23.1
 
 require (
-	github.com/aws/aws-sdk-go-v2/config v1.28.4
+	github.com/aws/aws-sdk-go-v2/config v1.28.6
 	github.com/grafana/sobek v0.0.0-20241024150027-d91f02b05e9b
 	github.com/linkedin/goavro/v2 v2.13.0
 	github.com/pavlo-v-chernykh/keystore-go/v4 v4.5.0
-	github.com/riferrei/srclient v0.7.0
+	github.com/riferrei/srclient v0.7.1
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1
 	github.com/segmentio/kafka-go v0.4.47
 	github.com/segmentio/kafka-go/sasl/aws_msk_iam_v2 v0.1.0
 	github.com/sirupsen/logrus v1.9.3
-	github.com/stretchr/testify v1.9.0
+	github.com/stretchr/testify v1.10.0
 	go.k6.io/k6 v0.55.0
 	gopkg.in/guregu/null.v3 v3.5.0
 )
 
 require (
-	github.com/aws/aws-sdk-go-v2 v1.32.4 // indirect
-	github.com/aws/aws-sdk-go-v2/credentials v1.17.45 // indirect
-	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.19 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.23 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.23 // indirect
+	github.com/aws/aws-sdk-go-v2 v1.32.6 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.17.47 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.21 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.25 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.25 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.0 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.4 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sso v1.24.5 // indirect
-	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.4 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sts v1.33.0 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.6 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.24.7 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.6 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.33.2 // indirect
 	github.com/aws/smithy-go v1.22.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
@@ -41,12 +41,12 @@ require (
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-sourcemap/sourcemap v2.1.4+incompatible // indirect
 	github.com/golang/snappy v0.0.4 // indirect
-	github.com/google/pprof v0.0.0-20241101162523-b92577c0c142 // indirect
+	github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad // indirect
 	github.com/google/uuid v1.6.0 // indirect
-	github.com/grpc-ecosystem/grpc-gateway/v2 v2.23.0 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.24.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/klauspost/compress v1.17.11 // indirect
-	github.com/mailru/easyjson v0.7.7 // indirect
+	github.com/mailru/easyjson v0.9.0 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mstoykov/atlas v0.0.0-20220811071828-388f114305dd // indirect
@@ -60,23 +60,25 @@ require (
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
-	go.opentelemetry.io/otel v1.32.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.32.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.32.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.32.0 // indirect
-	go.opentelemetry.io/otel/metric v1.32.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.32.0 // indirect
-	go.opentelemetry.io/otel/trace v1.32.0 // indirect
-	go.opentelemetry.io/proto/otlp v1.3.1 // indirect
-	golang.org/x/crypto v0.29.0 // indirect
-	golang.org/x/net v0.31.0 // indirect
-	golang.org/x/sync v0.9.0 // indirect
-	golang.org/x/sys v0.27.0 // indirect
-	golang.org/x/text v0.20.0 // indirect
+	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/otel v1.33.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.33.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.33.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.33.0 // indirect
+	go.opentelemetry.io/otel/metric v1.33.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.33.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.33.0 // indirect
+	go.opentelemetry.io/otel/trace v1.33.0 // indirect
+	go.opentelemetry.io/proto/otlp v1.4.0 // indirect
+	golang.org/x/crypto v0.31.0 // indirect
+	golang.org/x/net v0.32.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	golang.org/x/time v0.8.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20241113202542-65e8d215514f // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20241113202542-65e8d215514f // indirect
-	google.golang.org/grpc v1.68.0 // indirect
-	google.golang.org/protobuf v1.35.2 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20241216192217-9240e9c98484 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20241216192217-9240e9c98484 // indirect
+	google.golang.org/grpc v1.69.0 // indirect
+	google.golang.org/protobuf v1.36.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )