use long as timestamp type

xuxiong1 · xuxiong1 · commit c76a342ea211 · 2025-02-23T00:10:30.000+09:00
Signed-off-by: xuxiong1 &lt;xiongxug@outlook.com&gt;
diff --git a/plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaPartitionConsumer.java b/plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaPartitionConsumer.java
@@ -150,9 +150,8 @@ public IngestionShardPointer latestPointer() {
     }
 
     @Override
-    public IngestionShardPointer pointerFromTimestampMillis(String timestampMillisStr) {
+    public IngestionShardPointer pointerFromTimestampMillis(long timestampMillis) {
         long offset = AccessController.doPrivileged((PrivilegedAction<Long>) () -> {
-            long timestampMillis = Long.parseLong(timestampMillisStr);
             Map<TopicPartition, OffsetAndTimestamp> position = consumer.offsetsForTimes(
                 Collections.singletonMap(topicPartition, timestampMillis)
             );
@@ -166,7 +165,7 @@ public IngestionShardPointer pointerFromTimestampMillis(String timestampMillisSt
             return offsetAndTimestamp.offset();
         });
         if (offset < 0) {
-            logger.warn("No message found for timestamp {}, fall back to auto.offset.reset policy", timestampMillisStr);
+            logger.warn("No message found for timestamp {}, fall back to auto.offset.reset policy", timestampMillis);
             String autoOffsetResetConfig = config.getAutoOffsetResetConfig();
             if (OffsetResetStrategy.EARLIEST.toString().equals(autoOffsetResetConfig)) {
                 logger.warn("The auto.offset.reset is set to earliest, seek to earliest pointer");
@@ -175,7 +174,7 @@ public IngestionShardPointer pointerFromTimestampMillis(String timestampMillisSt
                 logger.warn("The auto.offset.reset is set to latest, seek to latest pointer");
                 return latestPointer();
             } else {
-                throw new IllegalArgumentException("No message found for timestamp " + timestampMillisStr);
+                throw new IllegalArgumentException("No message found for timestamp " + timestampMillis);
             }
         }
         return new KafkaOffset(offset);
diff --git a/plugins/ingestion-kafka/src/test/java/org/opensearch/plugin/kafka/KafkaPartitionConsumerTests.java b/plugins/ingestion-kafka/src/test/java/org/opensearch/plugin/kafka/KafkaPartitionConsumerTests.java
@@ -96,7 +96,7 @@ public void testPointerFromTimestampMillis() {
             Collections.singletonMap(topicPartition, new org.apache.kafka.clients.consumer.OffsetAndTimestamp(5L, 1000L))
         );
 
-        KafkaOffset offset = (KafkaOffset) consumer.pointerFromTimestampMillis("1000");
+        KafkaOffset offset = (KafkaOffset) consumer.pointerFromTimestampMillis(1000);
 
         assertEquals(5L, offset.getOffset());
     }
diff --git a/server/src/main/java/org/opensearch/index/IngestionShardConsumer.java b/server/src/main/java/org/opensearch/index/IngestionShardConsumer.java
@@ -88,7 +88,7 @@ public M getMessage() {
      * @param timestampMillis the timestamp in milliseconds
      * @return the ingestion shard pointer corresponding to the given timestamp
      */
-    IngestionShardPointer pointerFromTimestampMillis(String timestampMillis);
+    IngestionShardPointer pointerFromTimestampMillis(long timestampMillis);
 
     /**
      * Returns an ingestion shard pointer based on the provided offset.
diff --git a/server/src/main/java/org/opensearch/indices/pollingingest/DefaultStreamPoller.java b/server/src/main/java/org/opensearch/indices/pollingingest/DefaultStreamPoller.java
@@ -161,7 +161,7 @@ protected void startPoll() {
                             logger.info("Resetting offset by seeking to offset {}", batchStartPointer.asString());
                             break;
                         case REWIND_BY_TIMESTAMP:
-                            batchStartPointer = consumer.pointerFromTimestampMillis(resetValue);
+                            batchStartPointer = consumer.pointerFromTimestampMillis(Long.parseLong(resetValue));
                             logger.info(
                                 "Resetting offset by seeking to timestamp {}, corresponding offset {}",
                                 resetValue,
diff --git a/server/src/test/java/org/opensearch/index/engine/FakeIngestionSource.java b/server/src/test/java/org/opensearch/index/engine/FakeIngestionSource.java
@@ -94,7 +94,7 @@ public FakeIngestionShardPointer latestPointer() {
         }
 
         @Override
-        public IngestionShardPointer pointerFromTimestampMillis(String timestampMillis) {
+        public IngestionShardPointer pointerFromTimestampMillis(long timestampMillis) {
             throw new UnsupportedOperationException("Not implemented yet.");
         }
 

Original file line number	Diff line number	Diff line change
`@@ -96,7 +96,7 @@ public void testPointerFromTimestampMillis() {`
`96`	`96`	`Collections.singletonMap(topicPartition, new org.apache.kafka.clients.consumer.OffsetAndTimestamp(5L, 1000L))`
`97`	`97`	`);`
`98`	`98`
`99`		`- KafkaOffset offset = (KafkaOffset) consumer.pointerFromTimestampMillis("1000");`
	`99`	`+ KafkaOffset offset = (KafkaOffset) consumer.pointerFromTimestampMillis(1000);`
`100`	`100`
`101`	`101`	`assertEquals(5L, offset.getOffset());`
`102`	`102`	`}`
Original file line number	Diff line number	Diff line change
`@@ -94,7 +94,7 @@ public FakeIngestionShardPointer latestPointer() {`
`94`	`94`	`}`
`95`	`95`
`96`	`96`	`@Override`
`97`		`- public IngestionShardPointer pointerFromTimestampMillis(String timestampMillis) {`
	`97`	`+ public IngestionShardPointer pointerFromTimestampMillis(long timestampMillis) {`
`98`	`98`	`throw new UnsupportedOperationException("Not implemented yet.");`
`99`	`99`	`}`
`100`	`100`