Compute start and end offsets from timestamps

CHANGELOG.md

@@ -5,6 +5,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+### Changed
+- [#121](https://github.com/deviceinsight/kafkactl/issues/121) Support for from-timestamp message consumption
+
 
 ## 3.0.3 - 2023-02-01
 ### Changed

README.md

@@ -332,6 +332,18 @@ The consumer can be stopped when the latest offset is reached using `--exit` par
 kafkactl consume my-topic --from-beginning --exit
 ```
 
+The consumer can compute the offset it starts from using a timestamp :
+```bash
+kafkactl consume my-topic --from-timestamp <timestamp-with-milliseconds>
+```
+**NOTE:** `--from-timestamp` is not designed to schedule the beginning of consumer's consumption. The offset corresponding to the timestamp is computed at the begininng of the process. So if you set it to a date in the future, the consumer will start from the latest offset.
+
+The consumer can be stopped when the offset corresponding to a particuliar timestamp is reached :
+```bash
+kafkactl consume my-topic --from-timestamp <timestamp-with-milliseconds> --to-timestamp <timestamp-with-milliseconds>
+```
+**NOTE:** `--to-timestamp` is not designed to schedule the end of consumer's consumption. The offset corresponding to the timestamp is computed at the begininng of the process. So if you set it to a date in the future, the consumer will stop to the current latest offset.
+
 The following example prints keys in hex and values in base64:
 ```bash
 kafkactl consume my-topic --print-keys --key-encoding=hex --value-encoding=base64

cmd/consume/consume.go

@@ -33,6 +33,8 @@ func NewConsumeCmd() *cobra.Command {
 	cmdConsume.Flags().BoolVarP(&flags.PrintAvroSchema, "print-schema", "a", false, "print details about avro schema used for decoding")
 	cmdConsume.Flags().BoolVarP(&flags.PrintHeaders, "print-headers", "", false, "print message headers")
 	cmdConsume.Flags().IntVarP(&flags.Tail, "tail", "", -1, "show only the last n messages on the topic")
+	cmdConsume.Flags().Int64VarP(&flags.FromTimestamp, "from-timestamp", "", -1, "consume data from offset of given timestamp")
+	cmdConsume.Flags().Int64VarP(&flags.ToTimestamp, "to-timestamp", "", -1, "consume data till offset of given timestamp")
 	cmdConsume.Flags().Int64VarP(&flags.MaxMessages, "max-messages", "", -1, "stop consuming after n messages have been read")
 	cmdConsume.Flags().BoolVarP(&flags.Exit, "exit", "e", flags.Exit, "stop consuming when latest offset is reached")
 	cmdConsume.Flags().IntSliceVarP(&flags.Partitions, "partitions", "p", flags.Partitions, "partitions to consume. The default is to consume from all partitions.")

cmd/consume/consume_test.go

@@ -3,6 +3,7 @@ package consume_test
 import (
 	"encoding/hex"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -85,14 +86,102 @@ func TestConsumeTailIntegration(t *testing.T) {
 	messages := strings.Split(strings.TrimSpace(kafkaCtl.GetStdOut()), "\n")
 
 	if len(messages) != 3 {
-		t.Fatalf("expected 3 messages")
+		t.Fatalf("expected 3 messages, got %d", len(messages))
 	}
 
 	testutil.AssertEquals(t, "test-key-2#test-value-2a", messages[0])
 	testutil.AssertEquals(t, "test-key-3#test-value-3", messages[1])
 	testutil.AssertEquals(t, "test-key-2#test-value-2b", messages[2])
 }
 
+func TestConsumeFromTimestamp(t *testing.T) {
+
+	testutil.StartIntegrationTest(t)
+
+	topicName := testutil.CreateTopic(t, "consume-topic", "--partitions", "2")
+
+	testutil.ProduceMessageOnPartition(t, topicName, "key-1", "a", 0, 0)
+	testutil.ProduceMessageOnPartition(t, topicName, "key-1", "b", 0, 1)
+
+	time.Sleep(1 * time.Millisecond) // need to have messaged produced at different milliseconds to have reproducible test
+	t1 := time.Now().UnixMilli()
+
+	testutil.ProduceMessageOnPartition(t, topicName, "key-2", "c", 1, 0)
+	testutil.ProduceMessageOnPartition(t, topicName, "key-2", "d", 1, 1)
+	testutil.ProduceMessageOnPartition(t, topicName, "key-1", "e", 0, 2)
+	testutil.ProduceMessageOnPartition(t, topicName, "key-2", "f", 1, 2)
+
+	time.Sleep(1 * time.Millisecond)
+	t2 := time.Now().UnixMilli()
+
+	testutil.ProduceMessageOnPartition(t, topicName, "key-2", "g", 1, 3)
+	testutil.ProduceMessageOnPartition(t, topicName, "key-1", "h", 0, 3)
+
+	//test --from-timestamp with --to-timestamp
+	kafkaCtl := testutil.CreateKafkaCtlCommand()
+	if _, err := kafkaCtl.Execute("consume", topicName, "--from-timestamp", strconv.FormatInt(t1, 10), "--to-timestamp", strconv.FormatInt(t2, 10)); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+	messages := strings.Split(strings.TrimSpace(kafkaCtl.GetStdOut()), "\n")
+	testutil.AssertArraysEquals(t, []string{"c", "d", "e", "f"}, messages)
+
+	//test --from-timestamp with --max-messages (--partitions present for reproducibility)
+	kafkaCtl = testutil.CreateKafkaCtlCommand()
+	if _, err := kafkaCtl.Execute("consume", topicName, "--from-timestamp", strconv.FormatInt(t1, 10), "--max-messages", strconv.Itoa(2), "--partitions", strconv.Itoa(1)); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+	messages = strings.Split(strings.TrimSpace(kafkaCtl.GetStdOut()), "\n")
+	testutil.AssertArraysEquals(t, []string{"c", "d"}, messages)
+
+	//test --from-timestamp with --exit
+	kafkaCtl = testutil.CreateKafkaCtlCommand()
+	if _, err := kafkaCtl.Execute("consume", topicName, "--from-timestamp", strconv.FormatInt(t2, 10), "--exit"); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+	messages = strings.Split(strings.TrimSpace(kafkaCtl.GetStdOut()), "\n")
+	testutil.AssertArraysEquals(t, []string{"g", "h"}, messages)
+}
+
+func TestConsumeToTimestamp(t *testing.T) {
+
+	testutil.StartIntegrationTest(t)
+
+	topicName := testutil.CreateTopic(t, "consume-topic", "--partitions", "2")
+
+	testutil.ProduceMessageOnPartition(t, topicName, "key-1", "a", 0, 0)
+	testutil.ProduceMessageOnPartition(t, topicName, "key-1", "b", 0, 1)
+
+	time.Sleep(1 * time.Millisecond) // need to have messages produced at different milliseconds to have reproducible test
+	t1 := time.Now().UnixMilli()
+
+	testutil.ProduceMessageOnPartition(t, topicName, "key-2", "c", 1, 0)
+	testutil.ProduceMessageOnPartition(t, topicName, "key-2", "d", 1, 1)
+	testutil.ProduceMessageOnPartition(t, topicName, "key-1", "e", 0, 2)
+	testutil.ProduceMessageOnPartition(t, topicName, "key-2", "f", 1, 2)
+
+	time.Sleep(1 * time.Millisecond)
+	t2 := time.Now().UnixMilli()
+
+	testutil.ProduceMessageOnPartition(t, topicName, "key-2", "g", 1, 3)
+	testutil.ProduceMessageOnPartition(t, topicName, "key-1", "h", 0, 3)
+
+	//test --from-beginning with --to-timestamp
+	kafkaCtl := testutil.CreateKafkaCtlCommand()
+	if _, err := kafkaCtl.Execute("consume", topicName, "--from-beginning", "--to-timestamp", strconv.FormatInt(t1, 10)); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+	messages := strings.Split(strings.TrimSpace(kafkaCtl.GetStdOut()), "\n")
+	testutil.AssertArraysEquals(t, []string{"a", "b"}, messages)
+
+	//test --to-timestamp with --tail
+	kafkaCtl = testutil.CreateKafkaCtlCommand()
+	if _, err := kafkaCtl.Execute("consume", topicName, "--to-timestamp", strconv.FormatInt(t2, 10), "--tail", strconv.Itoa(4)); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+	messages = strings.Split(strings.TrimSpace(kafkaCtl.GetStdOut()), "\n")
+	testutil.AssertArraysEquals(t, []string{"c", "d", "e", "f"}, messages)
+}
+
 func TestConsumeWithKeyAndValueAsBase64Integration(t *testing.T) {
 
 	testutil.StartIntegrationTest(t)

internal/consume/PartitionConsumer.go

@@ -2,6 +2,7 @@ package consume
 
 import (
 	"context"
+	"math"
 	"strconv"
 	"strings"
 	"time"
@@ -12,6 +13,10 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
+const (
+	ErrOffset = math.MinInt64
+)
+
 type PartitionConsumer struct {
 	topic              string
 	partitions         []int32
@@ -69,7 +74,7 @@ func (c *PartitionConsumer) Start(ctx context.Context, flags Flags, messages cha
 				return errors.Errorf("Failed to start consumer for partition %d: %s", partitionID, err)
 			}
 
-			if lastOffset == -1 && (flags.Exit || flags.Tail > 0) {
+			if lastOffset == -1 && (flags.Exit || flags.Tail > 0 || flags.ToTimestamp > -1) {
 				output.Debugf("Skipping empty partition %d", partitionID)
 				return nil
 			} else if lastOffset == -1 || initialOffset <= lastOffset {
@@ -133,46 +138,70 @@ func (c *PartitionConsumer) Close() error {
 }
 
 func getOffsetBounds(client *sarama.Client, topic string, flags Flags, currentPartition int32) (int64, int64, error) {
-
-	if flags.Exit && len(flags.Offsets) == 0 && !flags.FromBeginning {
-		return -1, -1, errors.Errorf("parameter --exit has to be used in combination with --from-beginning or --offset")
-	} else if flags.Tail > 0 && len(flags.Offsets) > 0 {
-		return -1, -1, errors.Errorf("parameters --offset and --tail cannot be used together")
-	} else if flags.Tail > 0 {
-
-		newestOffset, oldestOffset, err := getBoundaryOffsets(client, topic, currentPartition)
-		if err != nil {
-			return -1, -1, err
-		}
-
-		minOffset := newestOffset - int64(flags.Tail)
-		maxOffset := newestOffset - 1
-		if minOffset < oldestOffset {
-			minOffset = oldestOffset
+	var startOffset int64
+	var endOffset int64
+	var err error
+	if startOffset, err = getStartOffset(client, topic, flags, currentPartition); err != nil {
+		return ErrOffset, ErrOffset, err
+	}
+	if endOffset, err = getEndOffset(client, topic, flags, currentPartition); err != nil {
+		return ErrOffset, ErrOffset, err
+	} else if startOffset == endOffset {
+		endOffset = -1 //nothing to consume on this partition
+	} else if endOffset != sarama.OffsetNewest {
+		endOffset = endOffset - 1
+	}
+	if flags.Tail > 0 && startOffset == sarama.OffsetNewest {
+		//When --tail is used compute startOffset so that it minimizes the number of messages consumed
+		if endOffset-int64(flags.Tail) > 0 {
+			startOffset = endOffset - int64(flags.Tail)
+		} else {
+			startOffset = sarama.OffsetOldest
 		}
-		return minOffset, maxOffset, nil
 	}
+	output.Debugf("consumer will consume offset %d to %d on partition %d", startOffset, endOffset, currentPartition)
+	return startOffset, endOffset, nil
+}
 
-	lastOffset := int64(-1)
-	oldestOffset := sarama.OffsetOldest
+func getStartOffset(client *sarama.Client, topic string, flags Flags, currentPartition int32) (int64, error) {
+	if hasExclusiveConditions(flags.FromTimestamp > -1, flags.FromBeginning, len(flags.Offsets) > 0) {
+		return -1, errors.Errorf("parameters '--from-timestamp', '--offset' and '--from-beginning' are exclusive")
+	}
+	if flags.FromTimestamp != -1 {
+		return (*client).GetOffset(topic, currentPartition, flags.FromTimestamp)
+	} else if flags.FromBeginning {
+		return (*client).GetOffset(topic, currentPartition, sarama.OffsetOldest)
+	} else if len(flags.Offsets) > 0 {
+		return extractOffsetForPartition(flags, currentPartition)
+	} else {
+		return sarama.OffsetNewest, nil
+	}
+}
 
-	if flags.Exit {
-		newestOffset, oldestOff, err := getBoundaryOffsets(client, topic, currentPartition)
-		if err != nil {
-			return -1, -1, err
+func getEndOffset(client *sarama.Client, topic string, flags Flags, currentPartition int32) (int64, error) {
+	if flags.ToTimestamp > -1 {
+		return (*client).GetOffset(topic, currentPartition, flags.ToTimestamp)
+	} else if flags.Exit || flags.Tail > 0 {
+		var newestOffset int64
+		var err error
+		if newestOffset, err = (*client).GetOffset(topic, currentPartition, sarama.OffsetNewest); err != nil {
+			return ErrOffset, err
 		}
-		lastOffset = newestOffset - 1
-		oldestOffset = oldestOff
+		return newestOffset, nil
+	} else {
+		return sarama.OffsetNewest, nil
 	}
+}
 
+func extractOffsetForPartition(flags Flags, currentPartition int32) (int64, error) {
 	for _, offsetFlag := range flags.Offsets {
 		offsetParts := strings.Split(offsetFlag, "=")
 
 		if len(offsetParts) == 2 {
 
 			partition, err := strconv.Atoi(offsetParts[0])
 			if err != nil {
-				return -1, -1, errors.Errorf("unable to parse offset parameter: %s (%v)", offsetFlag, err)
+				return -1, errors.Errorf("unable to parse offset parameter: %s (%v)", offsetFlag, err)
 			}
 
 			if int32(partition) != currentPartition {
@@ -181,27 +210,21 @@ func getOffsetBounds(client *sarama.Client, topic string, flags Flags, currentPa
 
 			offset, err := strconv.ParseInt(offsetParts[1], 10, 64)
 			if err != nil {
-				return -1, -1, errors.Errorf("unable to parse offset parameter: %s (%v)", offsetFlag, err)
+				return ErrOffset, errors.Errorf("unable to parse offset parameter: %s (%v)", offsetFlag, err)
 			}
 
-			return offset, lastOffset, nil
+			return offset, nil
 		}
 	}
-
-	if flags.FromBeginning {
-		return oldestOffset, lastOffset, nil
-	}
-	return sarama.OffsetNewest, -1, nil
+	return ErrOffset, errors.Errorf("unable to find offset parameter for partition %d: %s", currentPartition, flags.Offsets)
 }
 
-func getBoundaryOffsets(client *sarama.Client, topic string, partition int32) (newestOffset int64, oldestOffset int64, err error) {
-
-	if newestOffset, err = (*client).GetOffset(topic, partition, sarama.OffsetNewest); err != nil {
-		return -1, -1, errors.Errorf("failed to get offset for topic %s Partition %d: %v", topic, partition, err)
-	}
-
-	if oldestOffset, err = (*client).GetOffset(topic, partition, sarama.OffsetOldest); err != nil {
-		return -1, -1, errors.Errorf("failed to get offset for topic %s Partition %d: %v", topic, partition, err)
+func hasExclusiveConditions(flags ...bool) bool {
+	value := 0
+	for _, flag := range flags {
+		if flag {
+			value++
+		}
 	}
-	return newestOffset, oldestOffset, nil
+	return value > 1
 }

internal/consume/consume-operation.go

@@ -27,6 +27,8 @@ type Flags struct {
 	Partitions       []int
 	Offsets          []string
 	FromBeginning    bool
+	FromTimestamp    int64
+	ToTimestamp      int64
 	Tail             int
 	Exit             bool
 	MaxMessages      int64

testutil/helpers.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	"regexp"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -123,6 +124,17 @@ func ProduceMessage(t *testing.T, topic, key, value string, expectedPartition, e
 	AssertEquals(t, fmt.Sprintf("message produced (partition=%d\toffset=%d)", expectedPartition, expectedOffset), kafkaCtl.GetStdOut())
 }
 
+func ProduceMessageOnPartition(t *testing.T, topic, key, value string, partition int32, expectedOffset int64) {
+
+	kafkaCtl := CreateKafkaCtlCommand()
+
+	if _, err := kafkaCtl.Execute("produce", topic, "--key", key, "--value", value, "--partition", strconv.FormatInt(int64(partition), 10)); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+
+	AssertEquals(t, fmt.Sprintf("message produced (partition=%d\toffset=%d)", partition, expectedOffset), kafkaCtl.GetStdOut())
+}
+
 func VerifyGroupExists(t *testing.T, group string) {
 
 	kafkaCtl := CreateKafkaCtlCommand()

testutil/test_util.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"regexp"
+	"sort"
 	"strings"
 	"testing"
 	"time"
@@ -174,6 +175,15 @@ func AssertEquals(t *testing.T, expected, actual string) {
 	}
 }
 
+func AssertArraysEquals(t *testing.T, expected, actual []string) {
+	sort.Strings(expected)
+	sort.Strings(actual)
+
+	if !util.StrArrayEqual(actual, expected) {
+		t.Fatalf("unexpected values:\nexpected:\n--\n%s\n--\nactual:\n--\n%s\n--", expected, actual)
+	}
+}
+
 func AssertErrorContainsOneOf(t *testing.T, expected []string, err error) {
 
 	if err == nil {

util/util.go

@@ -17,3 +17,15 @@ func ContainsInt32(list []int32, element int32) bool {
 	}
 	return false
 }
+
+func StrArrayEqual(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i, v := range a {
+		if v != b[i] {
+			return false
+		}
+	}
+	return true
+}