Repository: authorjapps/zerocode (java)

Issue:
Feature Suggestion: Kafka Consumer consume from timestamp
Hi @authorjapps 
I think it would be a great feature if we could configure the kafka consumer to consume from a specific timestamp. If we had this, we would not have to cover backlog on the kafka topic if there is significant gap between consecutive test runs.

If you think this is a feature worth pursuing, I would love to contribute to the feature. We can have something like:

```
      "consumerLocalConfigs": {
          "seekTImestamp": "1706244382000"
      }

}
````

This would make the consumer seek to this timestamp and start consumption from there.

Pull request description:
ISSUE 615:  Changes to support "seekEpoch", "seekTimestamp" in Kafka Consumer step, removing "seek" from ConsumerCommonConfigs
# New Feature: Support to consume on all partitions from specified timestamp/epoch

PR Branch
https://github.com/a1shadows/zerocode/tree/issue-615/seekToTimestamp

fixes #615 

## Motivation and Context
Right now, the seek method available can only seek from one partition and has no semantics to consume from a specific timestamp. It only consumes the last 'n' messages on the specified partition.

With this new feature, we will be able to consume on all partitions from specified timestamp/epoch in the following fashion:

```
"consumerLocalConfigs" : {
    "seekTimestamp" : {
      "timestamp" : "2024-02-04T14:53:20.071",
      "format" : "yyyy-MM-dd'T'HH:mm:ss.SSS"
    }
```

or like this:

```
"consumerLocalConfigs" : {
    "seekEpoch" : "1707038600071"
  }
```    

in the kafka consume step.


In addition, we have also removed "seek" from ConsumerCommonConfigs as it made no sense in a common context and can now only be specified as a consumerLocalConfig.


## Checklist:

* [x] Unit tests added

* [x] Integration tests added

* [x] Test names are meaningful

* [x] Feature manually tested

* [x] Branch build passed

* [x] No 'package.*' in the imports

* [ ] Relevant Wiki page updated with clear instruction for the end user
  * [ ] Not applicable. This was only a refactor change, no functional or behaviour changes were introduced

* [ ] Http test added to `http-testing` module(if applicable) ?
  * [x] Not applicable. The changes did not affect HTTP automation flow

* [x] Kafka test added to `kafka-testing` module(if applicable) ?
  * [ ] Not applicable. The changes did not affect Kafka automation flow


Patch:
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigs.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigs.java
index 86599251..86b49df1 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigs.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigs.java
@@ -23,6 +23,9 @@ public class ConsumerLocalConfigs {
     private final String protoClassType;
     private final Boolean cacheByTopic;
     private final String filterByJsonPath;
+    private final String seekEpoch;
+    private final SeekTimestamp seekTimestamp;
+
 
     @JsonCreator
     public ConsumerLocalConfigs(
@@ -36,9 +39,11 @@ public class ConsumerLocalConfigs {
             @JsonProperty("pollingTime") Long pollingTime,
             @JsonProperty("cacheByTopic") Boolean cacheByTopic,
             @JsonProperty("filterByJsonPath") String filterByJsonPath,
-            @JsonProperty("seek") String seek) {
+            @JsonProperty("seek") String seek,
+            @JsonProperty("seekEpoch") String seekEpoch,
+            @JsonProperty("seekTimestamp") SeekTimestamp seekTimestamp) {
         this.recordType = recordType;
-        this.protoClassType= protobufMessageClassType;
+        this.protoClassType = protobufMessageClassType;
         this.fileDumpTo = fileDumpTo;
         this.commitAsync = commitAsync;
         this.commitSync = commitSync;
@@ -48,39 +53,45 @@ public class ConsumerLocalConfigs {
         this.cacheByTopic = cacheByTopic;
         this.filterByJsonPath = filterByJsonPath;
         this.seek = seek;
+        this.seekEpoch = seekEpoch;
+        this.seekTimestamp = seekTimestamp;
     }
-    
+
 
     public ConsumerLocalConfigs(
-             String recordType,
-             String fileDumpTo,
-             Boolean commitAsync,
+            String recordType,
+            String fileDumpTo,
+            Boolean commitAsync,
             Boolean commitSync,
             Boolean showRecordsConsumed,
             Integer maxNoOfRetryPollsOrTimeouts,
             Long pollingTime,
             Boolean cacheByTopic,
             String filterByJsonPath,
-            String seek) {
-		this(recordType, null,
+            String seek,
+            String seekEpoch,
+            SeekTimestamp seekTimestamp) {
+        this(recordType, null,
                 fileDumpTo,
                 commitAsync,
                 commitSync,
                 showRecordsConsumed,
                 maxNoOfRetryPollsOrTimeouts,
-				pollingTime,
+                pollingTime,
                 cacheByTopic,
                 filterByJsonPath,
-                seek);
+                seek,
+                seekEpoch,
+                seekTimestamp);
     }
 
     public String getRecordType() {
         return recordType != null ? recordType : RAW;
     }
-    
-	public String getProtoClassType() {
-		return protoClassType;
-	}
+
+    public String getProtoClassType() {
+        return protoClassType;
+    }
 
 
     public String getFileDumpTo() {
@@ -119,6 +130,14 @@ public class ConsumerLocalConfigs {
         return seek;
     }
 
+    public String getSeekEpoch() {
+        return seekEpoch;
+    }
+
+    public SeekTimestamp getSeekTimestamp() {
+        return seekTimestamp;
+    }
+
     @JsonIgnore
     public String[] getSeekTopicPartitionOffset() {
         return seek.split(",");
@@ -130,7 +149,7 @@ public class ConsumerLocalConfigs {
         if (o == null || getClass() != o.getClass()) return false;
         ConsumerLocalConfigs that = (ConsumerLocalConfigs) o;
         return Objects.equals(recordType, that.recordType) &&
-        		Objects.equals(protoClassType, that.protoClassType) &&
+                Objects.equals(protoClassType, that.protoClassType) &&
                 Objects.equals(fileDumpTo, that.fileDumpTo) &&
                 Objects.equals(commitAsync, that.commitAsync) &&
                 Objects.equals(commitSync, that.commitSync) &&
@@ -139,13 +158,14 @@ public class ConsumerLocalConfigs {
                 Objects.equals(pollingTime, that.pollingTime) &&
                 Objects.equals(filterByJsonPath, that.filterByJsonPath) &&
                 Objects.equals(cacheByTopic, that.cacheByTopic) &&
-                Objects.equals(seek, that.seek);
+                Objects.equals(seek, that.seek) &&
+                Objects.equals(seekEpoch, that.seekEpoch);
     }
 
     @Override
     public int hashCode() {
 
-        return Objects.hash(recordType, fileDumpTo, commitAsync, commitSync, showRecordsConsumed, maxNoOfRetryPollsOrTimeouts, pollingTime,cacheByTopic, filterByJsonPath, seek);
+        return Objects.hash(recordType, fileDumpTo, commitAsync, commitSync, showRecordsConsumed, maxNoOfRetryPollsOrTimeouts, pollingTime, cacheByTopic, filterByJsonPath, seek);
     }
 
     @Override
@@ -162,6 +182,8 @@ public class ConsumerLocalConfigs {
                 ", cacheByTopic=" + cacheByTopic +
                 ", filterByJsonPath=" + filterByJsonPath +
                 ", seek=" + seek +
+                ", seekEpoch=" + seekEpoch +
+                ", seekTimestamp=" + seekTimestamp +
                 '}';
     }
 }
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/consume/SeekTimestamp.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/consume/SeekTimestamp.java
new file mode 100644
index 00000000..6381ee25
--- /dev/null
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/consume/SeekTimestamp.java
@@ -0,0 +1,38 @@
+package org.jsmart.zerocode.core.kafka.consume;
+
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+@JsonInclude(JsonInclude.Include.NON_EMPTY)
+public class SeekTimestamp {
+
+    private final String timestamp;
+    private final String format;
+
+
+    @JsonCreator
+    public SeekTimestamp(
+            @JsonProperty("timestamp") String timestamp,
+            @JsonProperty("format") String format) {
+        this.timestamp = timestamp;
+        this.format = format;
+    }
+
+    public String getTimestamp() {
+        return timestamp;
+    }
+
+    public String getFormat() {
+        return format;
+    }
+
+    @Override
+    public String toString() {
+        return "SeekTimestamp{" +
+                "timestamp='" + timestamp + '\'' +
+                ", format='" + format + '\'' +
+                '}';
+    }
+}
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelper.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelper.java
index 1e5a76ae..0ea2d442 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelper.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelper.java
@@ -1,9 +1,32 @@
 package org.jsmart.zerocode.core.kafka.helper;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.io.Resources;
+import com.google.gson.Gson;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Message;
+import com.google.protobuf.MessageOrBuilder;
+import com.google.protobuf.util.JsonFormat;
+import com.jayway.jsonpath.JsonPath;
 import static java.lang.Integer.parseInt;
 import static java.lang.Long.parseLong;
 import static java.util.Optional.ofNullable;
 import static org.apache.commons.lang3.StringUtils.isEmpty;
+import static org.apache.commons.lang3.StringUtils.isNumeric;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+import org.jsmart.zerocode.core.di.provider.GsonSerDeProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
+import org.jsmart.zerocode.core.kafka.KafkaConstants;
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.AVRO;
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.DEFAULT_POLLING_TIME_MILLI_SEC;
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.JSON;
@@ -11,58 +34,44 @@ import static org.jsmart.zerocode.core.kafka.KafkaConstants.MAX_NO_OF_RETRY_POLL
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.PROTO;
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.RAW;
 import static org.jsmart.zerocode.core.kafka.common.KafkaCommonUtils.resolveValuePlaceHolders;
+import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigs;
+import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigsWrap;
+import org.jsmart.zerocode.core.kafka.consume.SeekTimestamp;
+import org.jsmart.zerocode.core.kafka.receive.ConsumerCommonConfigs;
+import org.jsmart.zerocode.core.kafka.receive.message.ConsumerJsonRecord;
+import org.jsmart.zerocode.core.kafka.receive.message.ConsumerJsonRecords;
+import org.jsmart.zerocode.core.kafka.receive.message.ConsumerRawRecords;
 import static org.jsmart.zerocode.core.utils.SmartUtils.prettyPrintJson;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.text.DateFormat;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 import java.util.Collections;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Properties;
 import java.util.Set;
-
-import com.jayway.jsonpath.JsonPath;
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.header.Header;
-import org.apache.kafka.common.header.Headers;
-import org.jsmart.zerocode.core.di.provider.GsonSerDeProvider;
-import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
-import org.jsmart.zerocode.core.kafka.KafkaConstants;
-import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigs;
-import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigsWrap;
-import org.jsmart.zerocode.core.kafka.receive.ConsumerCommonConfigs;
-import org.jsmart.zerocode.core.kafka.receive.message.ConsumerJsonRecord;
-import org.jsmart.zerocode.core.kafka.receive.message.ConsumerJsonRecords;
-import org.jsmart.zerocode.core.kafka.receive.message.ConsumerRawRecords;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.io.Resources;
-import com.google.gson.Gson;
-import com.google.protobuf.InvalidProtocolBufferException;
-import com.google.protobuf.Message;
-import com.google.protobuf.MessageOrBuilder;
-import com.google.protobuf.util.JsonFormat;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 public class KafkaConsumerHelper {
+    public static final String CONSUMER = "CONSUMER";
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConsumerHelper.class);
     private static final Gson gson = new GsonSerDeProvider().get();
     private static final ObjectMapper objectMapper = new ObjectMapperProvider().get();
-    public static final String CONSUMER = "CONSUMER";
     public static Map<String, Consumer> consumerCacheByTopicMap = new HashMap<>();
 
     public static Consumer createConsumer(String bootStrapServers, String consumerPropertyFile, String topic, Boolean consumerToBeCached) {
@@ -81,9 +90,9 @@ public class KafkaConsumerHelper {
             final Consumer consumer = new KafkaConsumer(properties);
             consumer.subscribe(Collections.singletonList(topic));
 
-            if(consumerToBeCached == true){
+            if (consumerToBeCached) {
                 consumerCacheByTopicMap.forEach((xTopic, xConsumer) -> {
-                    if(!xTopic.equals(topic)){
+                    if (!xTopic.equals(topic)) {
                         // close the earlier consumer if in the same group for safety.
                         // (even if not in the same group, closing it anyway will not do any harm)
                         // Otherwise rebalance will fail while rejoining/joining the same group for a new consumer
@@ -111,18 +120,18 @@ public class KafkaConsumerHelper {
 
     public static ConsumerRecords initialPollWaitingForConsumerGroupJoin(Consumer consumer, ConsumerLocalConfigs effectiveLocalConfigs) {
 
-            for (int run = 0; run < 50; run++) {
-                if (!consumer.assignment().isEmpty()) {
-                    LOGGER.debug("==> WaitingForConsumerGroupJoin - Partition now assigned. No records yet consumed");
-                    return new ConsumerRecords(new HashMap());
-                }
-                LOGGER.debug("==> WaitingForConsumerGroupJoin - Partition not assigned. Polling once");
-                ConsumerRecords records = consumer.poll(Duration.of(getPollTime(effectiveLocalConfigs), ChronoUnit.MILLIS));
-                LOGGER.debug("==> WaitingForConsumerGroupJoin - polled records length={}", records.count());
-                if (!records.isEmpty()) {
-                    return records;
-                }
+        for (int run = 0; run < 50; run++) {
+            if (!consumer.assignment().isEmpty()) {
+                LOGGER.debug("==> WaitingForConsumerGroupJoin - Partition now assigned. No records yet consumed");
+                return new ConsumerRecords(new HashMap());
+            }
+            LOGGER.debug("==> WaitingForConsumerGroupJoin - Partition not assigned. Polling once");
+            ConsumerRecords records = consumer.poll(Duration.of(getPollTime(effectiveLocalConfigs), ChronoUnit.MILLIS));
+            LOGGER.debug("==> WaitingForConsumerGroupJoin - polled records length={}", records.count());
+            if (!records.isEmpty()) {
+                return records;
             }
+        }
 
         throw new RuntimeException("\n********* Kafka Consumer unable to join in time - try increasing consumer polling time setting *********\n");
     }
@@ -135,6 +144,31 @@ public class KafkaConsumerHelper {
             validateCommitFlags(localCommitSync, localCommitAsync);
 
             validateSeekConfig(localConfigs);
+            validateSeekToTimestamp(localConfigs);
+        }
+    }
+
+    private static void validateSeekToTimestamp(ConsumerLocalConfigs localConfigs) {
+        String seekToTimestamp = localConfigs.getSeekEpoch();
+        if (isEmpty(seekToTimestamp)) {
+            if (isNumeric(seekToTimestamp) && (Long.parseLong(seekToTimestamp) > System.currentTimeMillis() || Long.parseLong(seekToTimestamp) < 0L)) {
+                throw new RuntimeException("\n------> 'seekEpoch' is not a valid epoch/Unix timestamp");
+            }
+            if (!isEmpty(localConfigs.getSeek()) && Objects.nonNull(localConfigs.getSeekTimestamp())) {
+                throw new RuntimeException("Only one of 'seek', 'seekEpoch' and 'seekTimestamp' should be provided, but not both. Please fix and rerun");
+            }
+        }
+        if (Objects.nonNull(localConfigs.getSeekTimestamp())) {
+            DateFormat dateFormat = new SimpleDateFormat(localConfigs.getSeekTimestamp().getFormat());
+            try {
+                Date date = dateFormat.parse(localConfigs.getSeekTimestamp().getTimestamp());
+                long epochMillis = date.toInstant().toEpochMilli();
+                if (epochMillis > System.currentTimeMillis() || epochMillis < 0L) {
+                    throw new RuntimeException("\n------> 'seekTimestamp' is not a valid epoch/Unix timestamp " + epochMillis);
+                }
+            } catch (ParseException e) {
+                throw new RuntimeException("Timestamp and format provided in 'seekTimestamp' cannot be parsed ", e);
+            }
         }
     }
 
@@ -163,7 +197,9 @@ public class KafkaConsumerHelper {
                     consumerCommon.getPollingTime(),
                     consumerCommon.getCacheByTopic(),
                     consumerCommon.getFilterByJsonPath(),
-                    consumerCommon.getSeek());
+                    null,
+                    null,
+                    null);
         }
 
         // Handle recordType
@@ -188,9 +224,6 @@ public class KafkaConsumerHelper {
         // Handle pollingTime
         String filterByJsonPath = ofNullable(consumerLocal.getFilterByJsonPath()).orElse(consumerCommon.getFilterByJsonPath());
 
-        // Handle pollingTime
-        String effectiveSeek = ofNullable(consumerLocal.getSeek()).orElse(consumerCommon.getSeek());
-
         // Handle consumerCache by topic
         Boolean effectiveConsumerCacheByTopic = ofNullable(consumerLocal.getCacheByTopic())
                 .orElse(consumerCommon.getCacheByTopic());
@@ -222,7 +255,9 @@ public class KafkaConsumerHelper {
                 effectivePollingTime,
                 effectiveConsumerCacheByTopic,
                 filterByJsonPath,
-                effectiveSeek);
+                consumerLocal.getSeek(),
+                consumerLocal.getSeekEpoch(),
+                consumerLocal.getSeekTimestamp());
     }
 
     public static ConsumerLocalConfigs readConsumerLocalTestProperties(String requestJsonWithConfigWrapped) {
@@ -264,7 +299,7 @@ public class KafkaConsumerHelper {
             Object key = thisRecord.key();
             Object valueObj = thisRecord.value();
             Headers headers = thisRecord.headers();
-            String keyStr =  thisRecord.key() != null ?  thisRecord.key().toString() : "";
+            String keyStr = thisRecord.key() != null ? thisRecord.key().toString() : "";
             String valueStr = consumerLocalConfig != null && KafkaConstants.PROTO.equalsIgnoreCase(consumerLocalConfig.getRecordType()) ? convertProtobufToJson(thisRecord, consumerLocalConfig) : valueObj.toString();
             LOGGER.debug("\nRecord Key - {} , Record value - {}, Record partition - {}, Record offset - {}, Headers - {}",
                     key, valueStr, thisRecord.partition(), thisRecord.offset(), headers);
@@ -289,7 +324,7 @@ public class KafkaConsumerHelper {
             throw new IllegalArgumentException(
                     "[protoClassType] is required consumer config for recordType PROTO.");
         }
-        MessageOrBuilder builderOrMessage = (MessageOrBuilder) createMessageOrBuilder(
+        MessageOrBuilder builderOrMessage = createMessageOrBuilder(
                 consumerLocalConfig.getProtoClassType(), (byte[]) thisRecord.value());
         try {
             return JsonFormat.printer().includingDefaultValueFields().preservingProtoFieldNames().print(builderOrMessage);
@@ -301,10 +336,10 @@ public class KafkaConsumerHelper {
     private static MessageOrBuilder createMessageOrBuilder(String messageClass, byte[] value) {
         try {
             Class<Message> msgClass = (Class<Message>) Class.forName(messageClass);
-            Method method = msgClass.getMethod("parseFrom", new Class[]{byte[].class});
+            Method method = msgClass.getMethod("parseFrom", byte[].class);
             return (MessageOrBuilder) method.invoke(null, value);
         } catch (IllegalAccessException | ClassNotFoundException | NoSuchMethodException | SecurityException
-                | IllegalArgumentException | InvocationTargetException e) {
+                 | IllegalArgumentException | InvocationTargetException e) {
             throw new IllegalArgumentException(e);
         }
 
@@ -316,7 +351,7 @@ public class KafkaConsumerHelper {
 
         String result;
 
-        if (testConfigs != null && testConfigs.getShowRecordsConsumed() == false) {
+        if (testConfigs != null && !testConfigs.getShowRecordsConsumed()) {
             int size = jsonRecords.size();
             result = prettyPrintJson(gson.toJson(new ConsumerRawRecords(size == 0 ? rawRecords.size() : size)));
 
@@ -326,7 +361,7 @@ public class KafkaConsumerHelper {
         } else if (testConfigs != null && (JSON.equals(testConfigs.getRecordType()) || PROTO.equalsIgnoreCase(testConfigs.getRecordType()) || AVRO.equalsIgnoreCase(testConfigs.getRecordType()))) {
             result = prettyPrintJson(objectMapper.writeValueAsString(new ConsumerJsonRecords(jsonRecords)));
 
-        }else {
+        } else {
             result = "{\"error\" : \"recordType Undecided, Please chose recordType as JSON or RAW\"}";
         }
 
@@ -363,10 +398,10 @@ public class KafkaConsumerHelper {
             effectiveCommitAsync = localCommitAsync;
         }
 
-        if (effectiveCommitSync != null && effectiveCommitSync == true) {
+        if (effectiveCommitSync != null && effectiveCommitSync) {
             consumer.commitSync();
 
-        } else if (effectiveCommitAsync != null && effectiveCommitAsync == true) {
+        } else if (effectiveCommitAsync != null && effectiveCommitAsync) {
             consumer.commitAsync();
 
         } else {
@@ -378,32 +413,78 @@ public class KafkaConsumerHelper {
         // --------------------------------------------------------
     }
 
-    public static void handleSeekOffset(ConsumerLocalConfigs effectiveLocal, Consumer consumer) {
+    public static void handleSeek(ConsumerLocalConfigs effectiveLocal, Consumer consumer, String topicName) {
         String seek = effectiveLocal.getSeek();
         if (!isEmpty(seek)) {
-            String[] seekParts = effectiveLocal.getSeekTopicPartitionOffset();
-            String topic = seekParts[0];
-            int partition = parseInt(seekParts[1]);
-            long offset = parseLong(seekParts[2]);
+            handleSeekByOffset(effectiveLocal, consumer);
+        } else if (!isEmpty(effectiveLocal.getSeekEpoch())) {
+            handleSeekByEpoch(Long.parseLong(effectiveLocal.getSeekEpoch()), consumer, topicName);
+        } else if (Objects.nonNull(effectiveLocal.getSeekTimestamp())) {
+            handleSeekByTimestamp(effectiveLocal.getSeekTimestamp(), consumer, topicName);
+        }
+    }
+
+    private static void handleSeekByTimestamp(SeekTimestamp seekTimestamp, Consumer consumer, String topicName) {
+        if (Objects.nonNull(seekTimestamp)) {
+            DateFormat dateFormat = new SimpleDateFormat(seekTimestamp.getFormat());
+            Date date = null;
+            try {
+                date = dateFormat.parse(seekTimestamp.getTimestamp());
+            } catch (ParseException e) {
+                throw new RuntimeException("Could not parse timestamp", e);
+            }
+            handleSeekByEpoch(date.toInstant().toEpochMilli(), consumer, topicName);
+        }
+    }
+
+    private static void handleSeekByEpoch(Long epoch, Consumer consumer, String topicName) {
+        if (Objects.nonNull(epoch)) {
+            List<PartitionInfo> partitionInfos = consumer.partitionsFor(topicName);
 
-            TopicPartition topicPartition = new TopicPartition(topic, partition);
-            Set<TopicPartition> topicPartitions = new HashSet<>();
-            topicPartitions.add(topicPartition);
+            //fetch partitions on topic
+            List<TopicPartition> topicPartitions = partitionInfos.stream()
+                    .map(info -> new TopicPartition(info.topic(), info.partition()))
+                    .collect(Collectors.toList());
 
+            //fetch offsets for each partition-timestamp pair
+            Map<TopicPartition, Long> topicPartitionTimestampMap = topicPartitions.stream()
+                    .collect(Collectors.toMap(Function.identity(), ignore -> epoch));
+            Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap = consumer.offsetsForTimes(topicPartitionTimestampMap);
+
+            //assign to fetched partitions
             consumer.unsubscribe();
-            consumer.assign(topicPartitions);
+            consumer.assign(topicPartitionOffsetAndTimestampMap.keySet());
 
-            if (offset <= -1) {
-                consumer.seekToEnd(topicPartitions);
-                consumer.seek(topicPartition, consumer.position(topicPartition) + offset);
-            } else {
-                consumer.seek(topicPartition, offset);
+            //seek to fetched offsets for partitions
+            for (Map.Entry<TopicPartition, OffsetAndTimestamp> topicOffsetEntry : topicPartitionOffsetAndTimestampMap.entrySet()) {
+                consumer.seek(topicOffsetEntry.getKey(), topicOffsetEntry.getValue().offset());
             }
         }
     }
 
+    private static void handleSeekByOffset(ConsumerLocalConfigs effectiveLocal, Consumer consumer) {
+        String[] seekParts = effectiveLocal.getSeekTopicPartitionOffset();
+        String topic = seekParts[0];
+        int partition = parseInt(seekParts[1]);
+        long offset = parseLong(seekParts[2]);
+
+        TopicPartition topicPartition = new TopicPartition(topic, partition);
+        Set<TopicPartition> topicPartitions = new HashSet<>();
+        topicPartitions.add(topicPartition);
+
+        consumer.unsubscribe();
+        consumer.assign(topicPartitions);
+
+        if (offset <= -1) {
+            consumer.seekToEnd(topicPartitions);
+            consumer.seek(topicPartition, consumer.position(topicPartition) + offset);
+        } else {
+            consumer.seek(topicPartition, offset);
+        }
+    }
+
     private static void validateCommitFlags(Boolean commitSync, Boolean commitAsync) {
-        if ((commitSync != null && commitAsync != null) && commitSync == true && commitAsync == true) {
+        if ((commitSync != null && commitAsync != null) && commitSync && commitAsync) {
             throw new RuntimeException("\n********* Both commitSync and commitAsync can not be true *********\n");
         }
     }
@@ -419,7 +500,7 @@ public class KafkaConsumerHelper {
     }
 
     private static Consumer getCachedConsumer(String topic, Boolean consumerToBeCached) {
-        if(consumerToBeCached){
+        if (consumerToBeCached) {
             return consumerCacheByTopicMap.get(topic);
         }
         return null;
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/ConsumerCommonConfigs.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/ConsumerCommonConfigs.java
index 2b6640d4..d34fe087 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/ConsumerCommonConfigs.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/ConsumerCommonConfigs.java
@@ -49,11 +49,6 @@ public class ConsumerCommonConfigs {
     @Named("consumer.filterByJsonPath")
     private String filterByJsonPath;
 
-    // TODO- Remove this from Global properties, as it doesn't make sense
-    @Inject(optional = true)
-    @Named("consumer.seek")
-    private String seek;
-
     public ConsumerCommonConfigs() {
     }
 
@@ -66,8 +61,7 @@ public class ConsumerCommonConfigs {
                                  Integer maxNoOfRetryPollsOrTimeouts,
                                  Long pollingTime,
                                  Boolean cacheByTopic,
-                                 String filterByJsonPath,
-                                 String seek
+                                 String filterByJsonPath
 
     ) {
         this.commitSync = commitSync;
@@ -80,7 +74,6 @@ public class ConsumerCommonConfigs {
         this.pollingTime = pollingTime;
         this.cacheByTopic = cacheByTopic;
         this.filterByJsonPath = filterByJsonPath;
-        this.seek = seek;
     }
     
     public ConsumerCommonConfigs(Boolean commitSync,
@@ -91,8 +84,7 @@ public class ConsumerCommonConfigs {
             Integer maxNoOfRetryPollsOrTimeouts,
             Long pollingTime,
             Boolean cacheByTopic,
-            String filterByJsonPath,
-            String seek
+            String filterByJsonPath
 
     ) {
 		this(commitSync,
@@ -104,8 +96,7 @@ public class ConsumerCommonConfigs {
                 maxNoOfRetryPollsOrTimeouts,
 				pollingTime,
                 cacheByTopic,
-                filterByJsonPath,
-                seek);
+                filterByJsonPath);
 	}
 
     public Boolean getCommitSync() {
@@ -136,9 +127,6 @@ public class ConsumerCommonConfigs {
         return recordType;
     }
 
-    public String getSeek() {
-        return seek;
-    }
 	public String getProtoClassType() {
 		return protoClassType;
 	}
@@ -165,7 +153,6 @@ public class ConsumerCommonConfigs {
                 ", pollingTime=" + pollingTime +
                 ", cacheByTopic=" + cacheByTopic +
                 ", filterByJsonPath=" + filterByJsonPath +
-                ", seek=" + seek +
                 '}';
     }
 }
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiver.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiver.java
index 37f89316..32725024 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiver.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiver.java
@@ -25,7 +25,7 @@ import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.deriveEf
 import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.getMaxTimeOuts;
 import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.getPollTime;
 import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.handleCommitSyncAsync;
-import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.handleSeekOffset;
+import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.handleSeek;
 import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.initialPollWaitingForConsumerGroupJoin;
 import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.prepareResult;
 import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.readConsumerLocalTestProperties;
@@ -62,7 +62,7 @@ public class KafkaReceiver {
 
         int noOfTimeOuts = 0;
 
-        handleSeekOffset(effectiveLocal, consumer);
+        handleSeek(effectiveLocal, consumer, topicName);
 
         LOGGER.debug("initial polling to trigger ConsumerGroupJoin");
 


Test patch:
diff --git a/core/src/test/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrapTest.java b/core/src/test/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrapTest.java
index 4a9de75d..63e2bf93 100644
--- a/core/src/test/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrapTest.java
+++ b/core/src/test/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrapTest.java
@@ -15,6 +15,86 @@ import static org.skyscreamer.jsonassert.JSONCompareMode.LENIENT;
 public class ConsumerLocalConfigsWrapTest {
     ObjectMapper objectMapper = new ObjectMapperProvider().get();
 
+    @Test
+    public void testSerDeser_seekEpoch() throws IOException {
+        ConsumerLocalConfigsWrap javaObject = new ConsumerLocalConfigsWrap(
+                new ConsumerLocalConfigs("RAW",
+                        "RAW:/target/ttt",
+                        true,
+                        null,
+                        true,
+                        3,
+                        50L,
+                        false,
+                        "$.JSON.Path",
+                        null,
+                        "1706940293669",
+                        null));
+        ObjectMapper objectMapper = new ObjectMapperProvider().get();
+
+        String json = objectMapper.writeValueAsString(javaObject);
+        assertEquals("{\n" +
+                        "    \"consumerLocalConfigs\":\n" +
+                        "    {\n" +
+                        "        \"recordType\": \"RAW\",\n" +
+                        "        \"fileDumpTo\": \"RAW:/target/ttt\",\n" +
+                        "        \"commitAsync\": true,\n" +
+                        "        \"showRecordsConsumed\": true,\n" +
+                        "        \"maxNoOfRetryPollsOrTimeouts\": 3,\n" +
+                        "        \"pollingTime\": 50,\n" +
+                        "        \"cacheByTopic\": false,\n" +
+                        "        \"filterByJsonPath\": \"$.JSON.Path\",\n" +
+                        "        \"seekEpoch\": \"1706940293669\"\n" +
+                        "    }\n" +
+                        "}",
+                json, LENIENT);
+
+        ConsumerLocalConfigsWrap javaPojo = objectMapper.readValue(json, ConsumerLocalConfigsWrap.class);
+        assertThat(javaPojo, is(javaObject));
+    }
+
+    @Test
+    public void testSerDeser_seekTimestamp() throws IOException {
+        ConsumerLocalConfigsWrap javaObject = new ConsumerLocalConfigsWrap(
+                new ConsumerLocalConfigs("RAW",
+                        "RAW:/target/ttt",
+                        true,
+                        null,
+                        true,
+                        3,
+                        50L,
+                        false,
+                        "$.JSON.Path",
+                        null,
+                        null,
+                        new SeekTimestamp("2024-01-29T19:35:21.959340", "yyyy-MM-dd'T'HH:mm:ss.ssssss")));
+        ObjectMapper objectMapper = new ObjectMapperProvider().get();
+
+        String json = objectMapper.writeValueAsString(javaObject);
+        assertEquals("{\n" +
+                        "    \"consumerLocalConfigs\":\n" +
+                        "    {\n" +
+                        "        \"recordType\": \"RAW\",\n" +
+                        "        \"fileDumpTo\": \"RAW:/target/ttt\",\n" +
+                        "        \"commitAsync\": true,\n" +
+                        "        \"showRecordsConsumed\": true,\n" +
+                        "        \"maxNoOfRetryPollsOrTimeouts\": 3,\n" +
+                        "        \"pollingTime\": 50,\n" +
+                        "        \"cacheByTopic\": false,\n" +
+                        "        \"filterByJsonPath\": \"$.JSON.Path\",\n" +
+                        "        \"seekTimestamp\":\n" +
+                        "        {\n" +
+                        "            \"timestamp\": \"2024-01-29T19:35:21.959340\",\n" +
+                        "            \"format\": \"yyyy-MM-dd'T'HH:mm:ss.ssssss\"\n" +
+                        "        }\n" +
+                        "    }\n" +
+                        "}",
+                json, LENIENT);
+
+        ConsumerLocalConfigsWrap javaPojo = objectMapper.readValue(json, ConsumerLocalConfigsWrap.class);
+        assertThat(javaPojo, is(javaObject));
+    }
+
     @Test
     public void testSerDeser() throws IOException {
         ConsumerLocalConfigsWrap javaObject = new ConsumerLocalConfigsWrap(
@@ -27,7 +107,8 @@ public class ConsumerLocalConfigsWrapTest {
                         50L,
                         false,
                         "$.JSON.Path",
-                        "1,0,test-topic"));
+                        "1,0,test-topic",
+                        null, null));
         ObjectMapper objectMapper = new ObjectMapperProvider().get();
 
         String json = objectMapper.writeValueAsString(javaObject);
@@ -58,7 +139,8 @@ public class ConsumerLocalConfigsWrapTest {
                         null,
                         false,
                         null,
-                        "1,0,test-topic"));
+                        "1,0,test-topic",
+                        null, null));
 
         String json = objectMapper.writeValueAsString(javaObject);
         assertEquals("{\n" +
diff --git a/core/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelperTest.java b/core/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelperTest.java
index e1b2eefd..63699055 100644
--- a/core/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelperTest.java
+++ b/core/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelperTest.java
@@ -44,7 +44,7 @@ public class KafkaConsumerHelperTest {
     public void test_syncAsyncTrueCommon() throws Exception {
         consumerCommon = new ConsumerCommonConfigs(true, true, "aTestFile",
                 "JSON", true, 3, 50L,
-                false,"$JSON.Path", "");
+                false,"$JSON.Path");
 
         expectedException.expectMessage("Both commitSync and commitAsync can not be true");
         ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(null, consumerCommon);
@@ -52,8 +52,8 @@ public class KafkaConsumerHelperTest {
 
     @Test
     public void test_syncAsyncTrueLocal() throws Exception {
-        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
-        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, true, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic");
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, true, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic", null, null);
         ConsumerLocalConfigsWrap localConfigsWrap = new ConsumerLocalConfigsWrap(consumerLocal);
 
         expectedException.expectMessage("Both commitSync and commitAsync can not be true");
@@ -63,8 +63,8 @@ public class KafkaConsumerHelperTest {
     @Test
     public void test_effectiveConfigsIsLocal() throws Exception {
 
-        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
-        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 150L,false,"$JSON.Path", "1,0,test-topic");
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 150L,false,"$JSON.Path", "1,0,test-topic", null, null);
 
         ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
 
@@ -79,7 +79,7 @@ public class KafkaConsumerHelperTest {
     @Test
     public void test_effectiveConfigsIsCentral() throws Exception {
 
-        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path");
         consumerLocal = null;
 
         ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
@@ -94,8 +94,8 @@ public class KafkaConsumerHelperTest {
     @Test
     public void test_effectiveCommitAsync_true() throws Exception {
 
-        consumerCommon = new ConsumerCommonConfigs(true, null, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
-        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic");
+        consumerCommon = new ConsumerCommonConfigs(true, null, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic", null, null);
 
         ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
 
@@ -106,8 +106,8 @@ public class KafkaConsumerHelperTest {
     @Test
     public void test_effectiveCommitSync_true() throws Exception {
 
-        consumerCommon = new ConsumerCommonConfigs(null, true, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
-        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", null, true, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic");
+        consumerCommon = new ConsumerCommonConfigs(null, true, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", null, true, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic", null, null);
 
         ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
 
@@ -118,8 +118,8 @@ public class KafkaConsumerHelperTest {
     @Test
     public void test_effectiveCommitSyncFromCommon_true() throws Exception {
 
-        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
-        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", null, null, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic");
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", null, null, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic", null, null);
 
         ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
 
@@ -130,8 +130,8 @@ public class KafkaConsumerHelperTest {
     @Test
     public void test_effectiveCommitAsyncFromCommon_true() throws Exception {
 
-        consumerCommon = new ConsumerCommonConfigs(null, true, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
-        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 150L,false,"$JSON.Path", "1,0,test-topic");
+        consumerCommon = new ConsumerCommonConfigs(null, true, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 150L,false,"$JSON.Path", "1,0,test-topic", null, null);
 
         ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
 
@@ -166,7 +166,7 @@ public class KafkaConsumerHelperTest {
     public void test_firstPoll_exits_early_on_assignment() {
 
         // given
-        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 1000L,false,"$JSON.Path", "");
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 1000L,false,"$JSON.Path");
         consumerLocal = null;
         ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
         Consumer consumer = Mockito.mock(Consumer.class);
@@ -185,8 +185,8 @@ public class KafkaConsumerHelperTest {
     public void test_firstPoll_exits_on_receiving_records() {
 
         // given
-        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 5000L,false,"$JSON.Path", "");
-        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic");
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 5000L,false,"$JSON.Path");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic", null, null);
         ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
         Consumer consumer = Mockito.mock(Consumer.class);
         Mockito.when(consumer.assignment()).thenReturn(new HashSet<TopicPartition>());
@@ -208,8 +208,8 @@ public class KafkaConsumerHelperTest {
     public void test_firstPoll_throws_after_timeout() throws Exception {
 
         // given
-        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, null,false,"$JSON.Path", "");
-        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic");
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, null,false,"$JSON.Path");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic", null, null);
         ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
 
         Consumer consumer = Mockito.mock(Consumer.class);
diff --git a/kafka-testing/src/main/java/org/jsmart/zerocode/zerocodejavaexec/utils/ExampleUtils.java b/kafka-testing/src/main/java/org/jsmart/zerocode/zerocodejavaexec/utils/ExampleUtils.java
index 86064618..90be7b93 100644
--- a/kafka-testing/src/main/java/org/jsmart/zerocode/zerocodejavaexec/utils/ExampleUtils.java
+++ b/kafka-testing/src/main/java/org/jsmart/zerocode/zerocodejavaexec/utils/ExampleUtils.java
@@ -1,5 +1,16 @@
 package org.jsmart.zerocode.zerocodejavaexec.utils;
 
+import org.jsmart.zerocode.core.kafka.consume.SeekTimestamp;
+
+import java.text.DateFormat;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+
 public class ExampleUtils {
 
+    public String seekTimestampToEpoch(SeekTimestamp seekTimestamp) throws ParseException {
+        DateFormat dateFormat = new SimpleDateFormat(seekTimestamp.getFormat());
+        return String.valueOf(dateFormat.parse(seekTimestamp.getTimestamp()).toInstant().toEpochMilli());
+    }
+
 }
diff --git a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaConsumeSeekOffsetTest.java b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaConsumeSeekOffsetTest.java
index bb48d0f1..690be202 100644
--- a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaConsumeSeekOffsetTest.java
+++ b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaConsumeSeekOffsetTest.java
@@ -26,4 +26,10 @@ public class KafkaConsumeSeekOffsetTest {
     public void testKafkaConsume_seekOffsetLatest() throws Exception {
     }
 
+    @Test
+    @Scenario("kafka/consume/test_kafka_consume_seek_epoch_and_timestamp.json")
+    public void testKafkaConsume_seekEpochAndTimestamp() {
+
+    }
+
 }
diff --git a/kafka-testing/src/test/resources/kafka/consume/test_kafka_consume_seek_epoch_and_timestamp.json b/kafka-testing/src/test/resources/kafka/consume/test_kafka_consume_seek_epoch_and_timestamp.json
new file mode 100755
index 00000000..b71aac92
--- /dev/null
+++ b/kafka-testing/src/test/resources/kafka/consume/test_kafka_consume_seek_epoch_and_timestamp.json
@@ -0,0 +1,105 @@
+{
+  "scenarioName": "Consume message after timestamp/epoch",
+  "steps": [
+    {
+      "name": "load_kafka_before_timestamp",
+      "url": "kafka-topic:demo-seekTime",
+      "operation": "PRODUCE",
+      "request": {
+        "records": [
+          {
+            "key": "${RANDOM.NUMBER}",
+            "value": "Before Timestamp 1"
+          },
+          {
+            "key": "${RANDOM.NUMBER}",
+            "value": "Before Timestamp 2"
+          }
+        ]
+      },
+      "assertions": {
+        "status": "Ok"
+      }
+    },
+    {
+      "name": "load_timestamp_and_epoch",
+      "url": "org.jsmart.zerocode.zerocodejavaexec.utils.ExampleUtils",
+      "operation": "seekTimestampToEpoch",
+      "request": {
+        "timestamp": "${LOCAL.DATETIME.NOW:yyyy-MM-dd'T'HH:mm:ss.SSS}",
+        "format": "yyyy-MM-dd'T'HH:mm:ss.SSS"
+      },
+      "assertions": {}
+    },
+    {
+      "name": "load_kafka_after_timestamp",
+      "url": "kafka-topic:demo-seekTime",
+      "operation": "PRODUCE",
+      "request": {
+        "records": [
+          {
+            "key": "${RANDOM.NUMBER}",
+            "value": "After Timestamp 1"
+          },
+          {
+            "key": "${RANDOM.NUMBER}",
+            "value": "After Timestamp 2"
+          }
+        ]
+      },
+      "assertions": {
+        "status": "Ok"
+      }
+    },
+    {
+      "name": "consume_seekEpoch",
+      "url": "kafka-topic:demo-seekTime",
+      "operation": "CONSUME",
+      "request": {
+        "consumerLocalConfigs": {
+          "seekEpoch": "${$.load_timestamp_and_epoch.response}",
+          "commitSync": true,
+          "recordType": "RAW",
+          "showRecordsConsumed": true,
+          "maxNoOfRetryPollsOrTimeouts": 3
+        }
+      },
+      "verify": {
+        "records": [
+          {
+            "value": "After Timestamp 1"
+          },
+          {
+            "value": "After Timestamp 2"
+          }
+        ]
+      }},
+    {
+      "name": "consume_seekTimestamp",
+      "url": "kafka-topic:demo-seekTime",
+      "operation": "CONSUME",
+      "request": {
+        "consumerLocalConfigs": {
+          "seekTimestamp": {
+            "timestamp": "${$.load_timestamp_and_epoch.request.timestamp}",
+            "format": "${$.load_timestamp_and_epoch.request.format}"
+          },
+          "commitSync": true,
+          "recordType": "RAW",
+          "showRecordsConsumed": true,
+          "maxNoOfRetryPollsOrTimeouts": 3
+        }
+      },
+      "verify": {
+        "records": [
+          {
+            "value": "After Timestamp 1"
+          },
+          {
+            "value": "After Timestamp 2"
+          }
+        ]
+      }
+    }
+  ]
+}