feat: implement time-based offset seeking across broker and client APIs

drmq-broker/src/main/java/com/drmq/broker/ClientHandler.java

@@ -89,6 +89,7 @@ private MessageEnvelope handleMessage(MessageEnvelope envelope) throws IOExcepti
             case PRE_VOTE_REQUEST -> handlePreVoteRequest(envelope);
             case APPEND_ENTRIES_REQUEST -> handleAppendEntriesRequest(envelope);
             case INSTALL_SNAPSHOT_REQUEST -> handleInstallSnapshotRequest(envelope);
+            case SEARCH_OFFSET_BY_TIME_REQUEST -> handleSearchOffsetByTimeRequest(envelope);
             default -> createErrorResponse("Unknown message type: " + envelope.getType());
         };
     }
@@ -480,6 +481,21 @@ private MessageEnvelope handleFetchOffsetRequest(MessageEnvelope envelope) throw
         }
     }
 
+    private MessageEnvelope handleSearchOffsetByTimeRequest(MessageEnvelope envelope) throws IOException {
+        SearchOffsetByTimeRequest request = SearchOffsetByTimeRequest.parseFrom(envelope.getPayload());
+
+        long offset = messageStore.findOffsetByTimestamp(request.getTopic(), request.getTimestamp());
+
+        SearchOffsetByTimeResponse.Builder response = SearchOffsetByTimeResponse.newBuilder()
+                .setSuccess(true)
+                .setOffset(offset);
+
+        return MessageEnvelope.newBuilder()
+                .setType(MessageType.SEARCH_OFFSET_BY_TIME_RESPONSE)
+                .setPayload(response.build().toByteString())
+                .build();
+    }
+
     @Deprecated
     private MessageEnvelope createErrorResponse(String errorMessage) {
         return createErrorResponse(errorMessage, MessageType.PRODUCE_RESPONSE);

drmq-broker/src/main/java/com/drmq/broker/MessageStore.java

@@ -306,6 +306,18 @@ public void lockForSnapshot(Runnable task) {
         }
     }
 
+    /**
+     * Get the first offset for a topic whose timestamp is >= targetTimestamp.
+     */
+    public long findOffsetByTimestamp(String topic, long targetTimestamp) {
+        try {
+            return logManager.findOffsetByTimestamp(topic, targetTimestamp);
+        } catch (IOException e) {
+            logger.error("Error searching for timestamp {} in topic {}", targetTimestamp, topic, e);
+            return -1;
+        }
+    }
+
     /**
      * Get a message by topic and offset.
      */

drmq-broker/src/main/java/com/drmq/broker/persistence/LogManager.java

@@ -119,6 +119,26 @@ public LogSegment getSegmentForOffset(String topic, long offset) {
         return null;
     }
 
+    /**
+     * Finds the first offset across all segments for a topic with timestamp >= targetTimestamp.
+     */
+    public long findOffsetByTimestamp(String topic, long targetTimestamp) throws IOException {
+        ConcurrentSkipListMap<Long, LogSegment> segments = topicSegments.get(topic);
+        if (segments == null || segments.isEmpty()) {
+            return -1;
+        }
+
+        // Iterate through segments to find the first one that has the timestamp.
+        for (LogSegment segment : segments.values()) {
+            long offset = segment.findOffsetByTimestamp(targetTimestamp);
+            if (offset != -1) {
+                return offset; // Found it in this segment!
+            }
+        }
+
+        return -1;
+    }
+
     public Map<String, ConcurrentSkipListMap<Long, LogSegment>> getAllSegments() {
         return topicSegments;
     }

drmq-broker/src/main/java/com/drmq/broker/persistence/LogSegment.java

@@ -225,6 +225,30 @@ public synchronized void truncate(long size) throws IOException {
         }
     }
 
+    /**
+     * Linearly scans the segment to find the first offset with timestamp >= targetTimestamp.
+     * @param targetTimestamp the timestamp to search for.
+     * @return the offset, or -1 if no message in this segment matches.
+     */
+    public synchronized long findOffsetByTimestamp(long targetTimestamp) throws IOException {
+        long position = 0;
+        long foundOffset = -1;
+        while (position < currentSize) {
+            try {
+                StoredMessage msg = read(position);
+                if (msg.getTimestamp() >= targetTimestamp) {
+                    foundOffset = msg.getOffset();
+                    break;
+                }
+                position += 4 + msg.getSerializedSize();
+            } catch (CorruptRecordException e) {
+                logger.warn("Corrupt record found while searching segment {} at pos {}", filePath, position);
+                break;
+            }
+        }
+        return foundOffset;
+    }
+
     @Override
     public void close() throws IOException {
         if (fileChannel.isOpen()) {

drmq-client/src/main/java/com/drmq/client/DRMQConsumer.java

@@ -315,6 +315,66 @@ public void subscribe(String topic, long fromOffset) throws IOException {
         logger.info("Subscribed to topic '{}' from explicit offset {} (group='{}')", topic, fromOffset, consumerGroup);
     }
 
+    /**
+     * Subscribe to a topic, starting from the first message at or after the given timestamp.
+     * Overrides the current offset for the topic.
+     */
+    public void seekByTime(String topic, long timestamp) throws IOException {
+        long offset = doSearchOffsetByTime(topic, timestamp);
+        if (offset >= 0) {
+            topicOffsets.put(topic, offset);
+            if (groupMode) {
+                commit(topic, offset);
+            }
+            logger.info("Subscribed to topic '{}' from time-based offset {} (timestamp={})", topic, offset, timestamp);
+        } else {
+            logger.warn("Could not find any offset for topic '{}' at or after timestamp {}", topic, timestamp);
+        }
+    }
+
+    private long doSearchOffsetByTime(String topic, long timestamp) throws IOException {
+        int attempts = 0;
+        while (attempts < MAX_RETRIES) {
+            try {
+                ensureConnectedWithRetry();
+                SearchOffsetByTimeRequest request = SearchOffsetByTimeRequest.newBuilder()
+                        .setTopic(topic)
+                        .setTimestamp(timestamp)
+                        .build();
+
+                sendEnvelope(MessageEnvelope.newBuilder()
+                        .setType(MessageType.SEARCH_OFFSET_BY_TIME_REQUEST)
+                        .setPayload(request.toByteString())
+                        .build());
+                MessageEnvelope responseEnvelope = receiveEnvelope();
+
+                if (responseEnvelope.getType() == MessageType.SEARCH_OFFSET_BY_TIME_RESPONSE) {
+                    SearchOffsetByTimeResponse resp = SearchOffsetByTimeResponse.parseFrom(responseEnvelope.getPayload());
+                    return resp.getOffset();
+                } else if (responseEnvelope.getType() == MessageType.PRODUCE_RESPONSE) {
+                    ProduceResponse errorResp = ProduceResponse.parseFrom(responseEnvelope.getPayload());
+                    if (tryRedirectToLeader(errorResp.getErrorMessage())) {
+                        attempts++;
+                        continue;
+                    }
+                    throw new IOException("Error searching offset by time: " + errorResp.getErrorMessage());
+                } else {
+                    throw new IOException("Unexpected response type: " + responseEnvelope.getType());
+                }
+            } catch (IOException e) {
+                if (e.getMessage() != null && e.getMessage().contains("Error searching offset")) {
+                    throw e;
+                }
+                attempts++;
+                if (attempts >= MAX_RETRIES) {
+                    throw new IOException("Failed to search offset by time after retries", e);
+                }
+                reconnect();
+            }
+        }
+        return -1;
+    }
+
     public List<ConsumedMessage> poll() throws IOException {
         return poll(DEFAULT_MAX_MESSAGES, DEFAULT_POLL_TIMEOUT_MS);
     }

drmq-client/src/main/java/com/drmq/client/commandLineExample/ConsumerApp.java

@@ -93,6 +93,36 @@ public static void main(String[] args) {
                         }
                     }
 
+                    case "seektime" -> {
+                        if (parts.length < 3) {
+                            System.out.println("❌ Usage: seektime <topic> <timestamp_or_iso_8601>");
+                            System.out.println("   Example: seektime orders 1783968079251");
+                            System.out.println("   Example: seektime payments 2026-07-13T10:30:00Z\n");
+                            continue;
+                        }
+
+                        String topic = parts[1];
+                        long targetTime;
+                        try {
+                            targetTime = Long.parseLong(parts[2]);
+                        } catch (NumberFormatException e) {
+                            try {
+                                targetTime = java.time.Instant.parse(parts[2]).toEpochMilli();
+                            } catch (Exception ex) {
+                                System.out.println("❌ Invalid timestamp format: " + parts[2]);
+                                System.out.println("   Please provide Unix milliseconds or an ISO-8601 UTC string (e.g., 2026-07-13T10:30:00Z)\n");
+                                continue;
+                            }
+                        }
+
+                        try {
+                            consumer.seekByTime(topic, targetTime);
+                            System.out.printf("✓ Seeked [%s] to messages on or after timestamp %d\n\n", topic, targetTime);
+                        } catch (IOException e) {
+                            System.out.printf("❌ Seek failed: %s\n\n", e.getMessage());
+                        }
+                    }
+
                     case "poll" -> {
                         int maxMessages = 100;
                         long timeoutMs = 1000; // default: 1s long-poll
@@ -273,6 +303,7 @@ private static void printHelp() {
         System.out.println("\nCommands:");
         System.out.println("  subscribe <topic>                    - Resume from broker-committed offset");
         System.out.println("  subscribe <topic> <offset>           - Override to a specific offset");
+        System.out.println("  seektime <topic> <timestamp>         - Seek to offset by Unix ms or ISO-8601 string");
         System.out.println("  poll [max] [timeout_ms]              - One-shot fetch (long-poll if timeout_ms>0)");
         System.out.println("    timeout_ms=0 → short poll (return immediately if empty)");
         System.out.println("    timeout_ms>0 → long poll (broker waits up to N ms for messages)");
@@ -287,6 +318,7 @@ private static void printHelp() {
         System.out.println("\nExamples:");
         System.out.println("  subscribe orders       (resume from last offset)");
         System.out.println("  subscribe payments 5   (seek to offset 5)");
+        System.out.println("  seektime orders 1783968079251 (seek by time)");
         System.out.println("  poll                   (long-poll, 1s timeout)");
         System.out.println("  poll 50 2000           (up to 50 msgs, wait 2s)");
         System.out.println("  poll 100 0             (short poll)");

drmq-integration-tests/src/test/java/com/drmq/integration/RaftIntegrationTest.java

@@ -332,4 +332,43 @@ void consumerReadsFromFollower() throws Exception {
             assertEquals("Message 2", messages.get(1).payloadAsString());
         }
     }
+
+    @Test
+    @DisplayName("Seek by time correctly replays from specific timestamp")
+    void consumerSeekByTimeTest() throws Exception {
+        startCluster();
+
+        BrokerServer leader = findLeader();
+        assertNotNull(leader);
+
+        String topic = "time-seek-test";
+
+        long targetTime = 0;
+        try (DRMQProducer producer = new DRMQProducer("localhost", leader.getPort())) {
+            producer.connect();
+            producer.send(topic, "Message 1 - Early").join();
+
+            Thread.sleep(100);
+            targetTime = System.currentTimeMillis();
+            Thread.sleep(100);
+
+            producer.send(topic, "Message 2 - Target").join();
+            producer.send(topic, "Message 3 - Late").join();
+        }
+
+        // Wait for replication
+        Thread.sleep(500);
+
+        // Single-mode consumer connecting to any node
+        BrokerServer follower = findFollower();
+        try (DRMQConsumer consumer = new DRMQConsumer("localhost", follower.getPort())) {
+            consumer.connect();
+            consumer.seekByTime(topic, targetTime);
+
+            var messages = consumer.poll(10, 2000);
+            assertEquals(2, messages.size(), "Should receive exactly messages 2 and 3");
+            assertEquals("Message 2 - Target", messages.get(0).payloadAsString());
+            assertEquals("Message 3 - Late", messages.get(1).payloadAsString());
+        }
+    }
 }

drmq-protocol/src/main/proto/messages.proto

@@ -98,6 +98,19 @@ message FetchOffsetResponse {
     string error_message = 3;
 }
 
+// Request to find an offset by a given timestamp
+message SearchOffsetByTimeRequest {
+    string topic = 1;
+    int64 timestamp = 2;      // The target timestamp
+}
+
+// Response with the found offset
+message SearchOffsetByTimeResponse {
+    bool success = 1;
+    int64 offset = 2;         // The found offset, or -1 if none
+    string error_message = 3;
+}
+
 // --- Dead-Letter Queue (DLQ) support ---
 
 // Request to negatively acknowledge (NACK) a message, triggering redelivery or DLQ routing
@@ -224,6 +237,8 @@ enum MessageType {
     NACK_RESPONSE = 19;
     PRODUCE_BATCH_REQUEST = 20;
     PRODUCE_BATCH_RESPONSE = 21;
+    SEARCH_OFFSET_BY_TIME_REQUEST = 22;
+    SEARCH_OFFSET_BY_TIME_RESPONSE = 23;
 }
 
 // --- Standard Error Codes ---

drmq-python-client/drmq_client.py

@@ -309,6 +309,45 @@ def subscribe(self, topic: str, from_offset: Optional[int] = None):
             else:
                 self.local_offsets[topic] = self._fetch_offset(topic)
 
+    def seek_by_time(self, topic: str, timestamp: int):
+        """
+        Subscribe to a topic, starting from the first message at or after the given timestamp.
+        Overrides the current offset for the topic.
+        """
+        if topic not in self.subscriptions:
+            self.subscriptions.append(topic)
+
+        for _ in range(self.max_retries):
+            try:
+                self._ensure_connected()
+                req = pb.SearchOffsetByTimeRequest()
+                req.topic = topic
+                req.timestamp = timestamp
+
+                resp_payload = self._send_envelope(pb.MessageType.SEARCH_OFFSET_BY_TIME_REQUEST, req.SerializeToString())
+                resp = pb.SearchOffsetByTimeResponse()
+                resp.ParseFromString(resp_payload)
+
+                if resp.success:
+                    if resp.offset >= 0:
+                        self.local_offsets[topic] = resp.offset
+                        if self.group_mode:
+                            self.commit(topic, resp.offset)
+                        logger.info(f"Subscribed to topic '{topic}' from time-based offset {resp.offset}")
+                    else:
+                        logger.warning(f"Could not find any offset for topic '{topic}' at or after timestamp {timestamp}")
+                    return
+                elif self._try_redirect_to_leader(resp.error_message):
+                    continue
+                else:
+                    raise DRMQConnectionError(f"Seek by time failed: {resp.error_message}")
+            except Exception as e:
+                if isinstance(e, DRMQConnectionError) and "Seek by time failed" in str(e):
+                    raise
+                self._reconnect()
+
+        raise DRMQConnectionError(f"Failed to seek offset by time after {self.max_retries} attempts")
+
     def poll(self, max_messages: int = 100, timeout_ms: int = 1000) -> List[pb.StoredMessage]:
         """Poll the broker for new messages across all subscribed topics."""
         for _ in range(self.max_retries):
@@ -342,6 +381,9 @@ def poll(self, max_messages: int = 100, timeout_ms: int = 1000) -> List[pb.Store
                     elif self._try_redirect_to_leader(resp.error_message):
                         # Break and retry the entire poll loop on the new leader
                         raise ConnectionError("Redirected to leader")
+                    else:
+                        print(f"POLL ERROR from broker: {resp.error_message}")
+                        raise DRMQConnectionError(f"Poll failed: {resp.error_message}")
 
                 return all_messages
             except Exception: