diff --git a/drmq-broker/src/main/java/com/drmq/broker/ClientHandler.java b/drmq-broker/src/main/java/com/drmq/broker/ClientHandler.java index 7e12499..ea03028 100644 --- a/drmq-broker/src/main/java/com/drmq/broker/ClientHandler.java +++ b/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); diff --git a/drmq-broker/src/main/java/com/drmq/broker/MessageStore.java b/drmq-broker/src/main/java/com/drmq/broker/MessageStore.java index 66e68b9..637ae54 100644 --- a/drmq-broker/src/main/java/com/drmq/broker/MessageStore.java +++ b/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. */ diff --git a/drmq-broker/src/main/java/com/drmq/broker/persistence/LogManager.java b/drmq-broker/src/main/java/com/drmq/broker/persistence/LogManager.java index d0822d7..3c1b401 100644 --- a/drmq-broker/src/main/java/com/drmq/broker/persistence/LogManager.java +++ b/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 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> getAllSegments() { return topicSegments; } diff --git a/drmq-broker/src/main/java/com/drmq/broker/persistence/LogSegment.java b/drmq-broker/src/main/java/com/drmq/broker/persistence/LogSegment.java index 3c3297c..9f15011 100644 --- a/drmq-broker/src/main/java/com/drmq/broker/persistence/LogSegment.java +++ b/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()) { diff --git a/drmq-client/src/main/java/com/drmq/client/DRMQConsumer.java b/drmq-client/src/main/java/com/drmq/client/DRMQConsumer.java index 750751f..87e7260 100644 --- a/drmq-client/src/main/java/com/drmq/client/DRMQConsumer.java +++ b/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 poll() throws IOException { return poll(DEFAULT_MAX_MESSAGES, DEFAULT_POLL_TIMEOUT_MS); } diff --git a/drmq-client/src/main/java/com/drmq/client/commandLineExample/ConsumerApp.java b/drmq-client/src/main/java/com/drmq/client/commandLineExample/ConsumerApp.java index 0950a3c..df7e533 100644 --- a/drmq-client/src/main/java/com/drmq/client/commandLineExample/ConsumerApp.java +++ b/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 "); + 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 - Resume from broker-committed offset"); System.out.println(" subscribe - Override to a specific offset"); + System.out.println(" seektime - 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)"); diff --git a/drmq-integration-tests/src/test/java/com/drmq/integration/RaftIntegrationTest.java b/drmq-integration-tests/src/test/java/com/drmq/integration/RaftIntegrationTest.java index 03c59d5..25b1676 100644 --- a/drmq-integration-tests/src/test/java/com/drmq/integration/RaftIntegrationTest.java +++ b/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()); + } + } } diff --git a/drmq-protocol/src/main/proto/messages.proto b/drmq-protocol/src/main/proto/messages.proto index a1a3df1..82713d0 100644 --- a/drmq-protocol/src/main/proto/messages.proto +++ b/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 --- diff --git a/drmq-python-client/drmq_client.py b/drmq-python-client/drmq_client.py index fea59b2..baf0460 100644 --- a/drmq-python-client/drmq_client.py +++ b/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: diff --git a/drmq-python-client/messages_pb2.py b/drmq-python-client/messages_pb2.py index 97ce045..310ffe7 100644 --- a/drmq-python-client/messages_pb2.py +++ b/drmq-python-client/messages_pb2.py @@ -1,11 +1,22 @@ # -*- coding: utf-8 -*- # Generated by the protocol buffer compiler. DO NOT EDIT! +# NO CHECKED-IN PROTOBUF GENCODE # source: messages.proto +# Protobuf Python Version: 7.35.0 """Generated protocol buffer code.""" -from google.protobuf.internal import builder as _builder from google.protobuf import descriptor as _descriptor from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import runtime_version as _runtime_version from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder +_runtime_version.ValidateProtobufRuntimeVersion( + _runtime_version.Domain.PUBLIC, + 7, + 35, + 0, + '', + 'messages.proto' +) # @@protoc_insertion_point(imports) _sym_db = _symbol_database.Default() @@ -13,64 +24,70 @@ -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0emessages.proto\x12\x11\x63om.drmq.protocol\"]\n\x0eProduceRequest\x12\r\n\x05topic\x18\x01 \x01(\t\x12\x0f\n\x07payload\x18\x02 \x01(\x0c\x12\x10\n\x03key\x18\x03 \x01(\tH\x00\x88\x01\x01\x12\x11\n\ttimestamp\x18\x04 \x01(\x03\x42\x06\n\x04_key\"I\n\x0fProduceResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x0e\n\x06offset\x18\x02 \x01(\x03\x12\x15\n\rerror_message\x18\x03 \x01(\t\"\xbb\x01\n\x13ProduceBatchRequest\x12\r\n\x05topic\x18\x01 \x01(\t\x12\x42\n\x07\x65ntries\x18\x02 \x03(\x0b\x32\x31.com.drmq.protocol.ProduceBatchRequest.BatchEntry\x1aQ\n\nBatchEntry\x12\x0f\n\x07payload\x18\x01 \x01(\x0c\x12\x10\n\x03key\x18\x02 \x01(\tH\x00\x88\x01\x01\x12\x18\n\x10\x63lient_timestamp\x18\x03 \x01(\x03\x42\x06\n\x04_key\"b\n\x14ProduceBatchResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x13\n\x0b\x62\x61se_offset\x18\x02 \x01(\x03\x12\r\n\x05\x63ount\x18\x03 \x01(\x05\x12\x15\n\rerror_message\x18\x04 \x01(\t\"\x7f\n\rStoredMessage\x12\x0e\n\x06offset\x18\x01 \x01(\x03\x12\r\n\x05topic\x18\x02 \x01(\t\x12\x0f\n\x07payload\x18\x03 \x01(\x0c\x12\x10\n\x03key\x18\x04 \x01(\tH\x00\x88\x01\x01\x12\x11\n\ttimestamp\x18\x05 \x01(\x03\x12\x11\n\tstored_at\x18\x06 \x01(\x03\x42\x06\n\x04_key\"\xb8\x01\n\x0e\x43onsumeRequest\x12\r\n\x05topic\x18\x01 \x01(\t\x12\x13\n\x0b\x66rom_offset\x18\x02 \x01(\x03\x12\x14\n\x0cmax_messages\x18\x03 \x01(\x05\x12\x12\n\ntimeout_ms\x18\x04 \x01(\x03\x12\x1b\n\x0e\x63onsumer_group\x18\x05 \x01(\tH\x00\x88\x01\x01\x12\x18\n\x0b\x63onsumer_id\x18\x06 \x01(\tH\x01\x88\x01\x01\x42\x11\n\x0f_consumer_groupB\x0e\n\x0c_consumer_id\"m\n\x0f\x43onsumeResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x32\n\x08messages\x18\x02 \x03(\x0b\x32 .com.drmq.protocol.StoredMessage\x12\x15\n\rerror_message\x18\x03 \x01(\t\"v\n\x13\x43ommitOffsetRequest\x12\x16\n\x0e\x63onsumer_group\x18\x01 \x01(\t\x12\r\n\x05topic\x18\x02 \x01(\t\x12\x0e\n\x06offset\x18\x03 \x01(\x03\x12\x18\n\x0b\x63onsumer_id\x18\x04 \x01(\tH\x00\x88\x01\x01\x42\x0e\n\x0c_consumer_id\">\n\x14\x43ommitOffsetResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x15\n\rerror_message\x18\x02 \x01(\t\";\n\x12\x46\x65tchOffsetRequest\x12\x16\n\x0e\x63onsumer_group\x18\x01 \x01(\t\x12\r\n\x05topic\x18\x02 \x01(\t\"M\n\x13\x46\x65tchOffsetResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x0e\n\x06offset\x18\x02 \x01(\x03\x12\x15\n\rerror_message\x18\x03 \x01(\t\"n\n\x0bNackRequest\x12\x16\n\x0e\x63onsumer_group\x18\x01 \x01(\t\x12\r\n\x05topic\x18\x02 \x01(\t\x12\x0e\n\x06offset\x18\x03 \x01(\x03\x12\x18\n\x0b\x63onsumer_id\x18\x04 \x01(\tH\x00\x88\x01\x01\x42\x0e\n\x0c_consumer_id\"M\n\x0cNackResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x15\n\rrouted_to_dlq\x18\x02 \x01(\x08\x12\x15\n\rerror_message\x18\x03 \x01(\t\"P\n\x0fMessageEnvelope\x12,\n\x04type\x18\x01 \x01(\x0e\x32\x1e.com.drmq.protocol.MessageType\x12\x0f\n\x07payload\x18\x02 \x01(\x0c\"\x8b\x02\n\tRaftEntry\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\r\n\x05index\x18\x02 \x01(\x03\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x0f\n\x07payload\x18\x04 \x01(\x0c\x12\x10\n\x03key\x18\x05 \x01(\tH\x00\x88\x01\x01\x12\x11\n\ttimestamp\x18\x06 \x01(\x03\x12\x38\n\x0c\x63ommand_type\x18\x07 \x01(\x0e\x32\".com.drmq.protocol.RaftCommandType\x12\x1b\n\x0e\x63onsumer_group\x18\x08 \x01(\tH\x01\x88\x01\x01\x12\x19\n\x0coffset_value\x18\t \x01(\x03H\x02\x88\x01\x01\x42\x06\n\x04_keyB\x11\n\x0f_consumer_groupB\x0f\n\r_offset_value\"g\n\x12RequestVoteRequest\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x14\n\x0c\x63\x61ndidate_id\x18\x02 \x01(\t\x12\x16\n\x0elast_log_index\x18\x03 \x01(\x03\x12\x15\n\rlast_log_term\x18\x04 \x01(\x03\"9\n\x13RequestVoteResponse\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x14\n\x0cvote_granted\x18\x02 \x01(\x08\"c\n\x0ePreVoteRequest\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x14\n\x0c\x63\x61ndidate_id\x18\x02 \x01(\t\x12\x16\n\x0elast_log_index\x18\x03 \x01(\x03\x12\x15\n\rlast_log_term\x18\x04 \x01(\x03\"5\n\x0fPreVoteResponse\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x14\n\x0cvote_granted\x18\x02 \x01(\x08\"\xac\x01\n\x14\x41ppendEntriesRequest\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x11\n\tleader_id\x18\x02 \x01(\t\x12\x16\n\x0eprev_log_index\x18\x03 \x01(\x03\x12\x15\n\rprev_log_term\x18\x04 \x01(\x03\x12-\n\x07\x65ntries\x18\x05 \x03(\x0b\x32\x1c.com.drmq.protocol.RaftEntry\x12\x15\n\rleader_commit\x18\x06 \x01(\x03\"K\n\x15\x41ppendEntriesResponse\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x0f\n\x07success\x18\x02 \x01(\x08\x12\x13\n\x0bmatch_index\x18\x03 \x01(\x03\"\x9e\x01\n\x16InstallSnapshotRequest\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x11\n\tleader_id\x18\x02 \x01(\t\x12\x1b\n\x13last_included_index\x18\x03 \x01(\x03\x12\x1a\n\x12last_included_term\x18\x04 \x01(\x03\x12\x0e\n\x06offset\x18\x05 \x01(\x03\x12\x0c\n\x04\x64\x61ta\x18\x06 \x01(\x0c\x12\x0c\n\x04\x64one\x18\x07 \x01(\x08\"\'\n\x17InstallSnapshotResponse\x12\x0c\n\x04term\x18\x01 \x01(\x03*D\n\x0fRaftCommandType\x12\x0b\n\x07MESSAGE\x10\x00\x12\x11\n\rOFFSET_COMMIT\x10\x01\x12\x11\n\rBATCH_MESSAGE\x10\x02*\x9f\x04\n\x0bMessageType\x12\x0b\n\x07UNKNOWN\x10\x00\x12\x13\n\x0fPRODUCE_REQUEST\x10\x01\x12\x14\n\x10PRODUCE_RESPONSE\x10\x02\x12\x13\n\x0f\x43ONSUME_REQUEST\x10\x03\x12\x14\n\x10\x43ONSUME_RESPONSE\x10\x04\x12\r\n\tHEARTBEAT\x10\x05\x12\x19\n\x15\x43OMMIT_OFFSET_REQUEST\x10\x06\x12\x1a\n\x16\x43OMMIT_OFFSET_RESPONSE\x10\x07\x12\x18\n\x14\x46\x45TCH_OFFSET_REQUEST\x10\x08\x12\x19\n\x15\x46\x45TCH_OFFSET_RESPONSE\x10\t\x12\x18\n\x14REQUEST_VOTE_REQUEST\x10\n\x12\x19\n\x15REQUEST_VOTE_RESPONSE\x10\x0b\x12\x1a\n\x16\x41PPEND_ENTRIES_REQUEST\x10\x0c\x12\x1b\n\x17\x41PPEND_ENTRIES_RESPONSE\x10\r\x12\x14\n\x10PRE_VOTE_REQUEST\x10\x0e\x12\x15\n\x11PRE_VOTE_RESPONSE\x10\x0f\x12\x1c\n\x18INSTALL_SNAPSHOT_REQUEST\x10\x10\x12\x1d\n\x19INSTALL_SNAPSHOT_RESPONSE\x10\x11\x12\x10\n\x0cNACK_REQUEST\x10\x12\x12\x11\n\rNACK_RESPONSE\x10\x13\x12\x19\n\x15PRODUCE_BATCH_REQUEST\x10\x14\x12\x1a\n\x16PRODUCE_BATCH_RESPONSE\x10\x15\x42#\n\x11\x63om.drmq.protocolB\x0c\x44RMQProtocolP\x00\x62\x06proto3') - -_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals()) -_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'messages_pb2', globals()) -if _descriptor._USE_C_DESCRIPTORS == False: +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0emessages.proto\x12\x11\x63om.drmq.protocol\"]\n\x0eProduceRequest\x12\r\n\x05topic\x18\x01 \x01(\t\x12\x0f\n\x07payload\x18\x02 \x01(\x0c\x12\x10\n\x03key\x18\x03 \x01(\tH\x00\x88\x01\x01\x12\x11\n\ttimestamp\x18\x04 \x01(\x03\x42\x06\n\x04_key\"{\n\x0fProduceResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x0e\n\x06offset\x18\x02 \x01(\x03\x12\x15\n\rerror_message\x18\x03 \x01(\t\x12\x30\n\nerror_code\x18\x04 \x01(\x0e\x32\x1c.com.drmq.protocol.ErrorCode\"\xbb\x01\n\x13ProduceBatchRequest\x12\r\n\x05topic\x18\x01 \x01(\t\x12\x42\n\x07\x65ntries\x18\x02 \x03(\x0b\x32\x31.com.drmq.protocol.ProduceBatchRequest.BatchEntry\x1aQ\n\nBatchEntry\x12\x0f\n\x07payload\x18\x01 \x01(\x0c\x12\x10\n\x03key\x18\x02 \x01(\tH\x00\x88\x01\x01\x12\x18\n\x10\x63lient_timestamp\x18\x03 \x01(\x03\x42\x06\n\x04_key\"\x94\x01\n\x14ProduceBatchResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x13\n\x0b\x62\x61se_offset\x18\x02 \x01(\x03\x12\r\n\x05\x63ount\x18\x03 \x01(\x05\x12\x15\n\rerror_message\x18\x04 \x01(\t\x12\x30\n\nerror_code\x18\x05 \x01(\x0e\x32\x1c.com.drmq.protocol.ErrorCode\"\x7f\n\rStoredMessage\x12\x0e\n\x06offset\x18\x01 \x01(\x03\x12\r\n\x05topic\x18\x02 \x01(\t\x12\x0f\n\x07payload\x18\x03 \x01(\x0c\x12\x10\n\x03key\x18\x04 \x01(\tH\x00\x88\x01\x01\x12\x11\n\ttimestamp\x18\x05 \x01(\x03\x12\x11\n\tstored_at\x18\x06 \x01(\x03\x42\x06\n\x04_key\"\xb8\x01\n\x0e\x43onsumeRequest\x12\r\n\x05topic\x18\x01 \x01(\t\x12\x13\n\x0b\x66rom_offset\x18\x02 \x01(\x03\x12\x14\n\x0cmax_messages\x18\x03 \x01(\x05\x12\x12\n\ntimeout_ms\x18\x04 \x01(\x03\x12\x1b\n\x0e\x63onsumer_group\x18\x05 \x01(\tH\x00\x88\x01\x01\x12\x18\n\x0b\x63onsumer_id\x18\x06 \x01(\tH\x01\x88\x01\x01\x42\x11\n\x0f_consumer_groupB\x0e\n\x0c_consumer_id\"m\n\x0f\x43onsumeResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x32\n\x08messages\x18\x02 \x03(\x0b\x32 .com.drmq.protocol.StoredMessage\x12\x15\n\rerror_message\x18\x03 \x01(\t\"v\n\x13\x43ommitOffsetRequest\x12\x16\n\x0e\x63onsumer_group\x18\x01 \x01(\t\x12\r\n\x05topic\x18\x02 \x01(\t\x12\x0e\n\x06offset\x18\x03 \x01(\x03\x12\x18\n\x0b\x63onsumer_id\x18\x04 \x01(\tH\x00\x88\x01\x01\x42\x0e\n\x0c_consumer_id\">\n\x14\x43ommitOffsetResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x15\n\rerror_message\x18\x02 \x01(\t\";\n\x12\x46\x65tchOffsetRequest\x12\x16\n\x0e\x63onsumer_group\x18\x01 \x01(\t\x12\r\n\x05topic\x18\x02 \x01(\t\"M\n\x13\x46\x65tchOffsetResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x0e\n\x06offset\x18\x02 \x01(\x03\x12\x15\n\rerror_message\x18\x03 \x01(\t\"=\n\x19SearchOffsetByTimeRequest\x12\r\n\x05topic\x18\x01 \x01(\t\x12\x11\n\ttimestamp\x18\x02 \x01(\x03\"T\n\x1aSearchOffsetByTimeResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x0e\n\x06offset\x18\x02 \x01(\x03\x12\x15\n\rerror_message\x18\x03 \x01(\t\"n\n\x0bNackRequest\x12\x16\n\x0e\x63onsumer_group\x18\x01 \x01(\t\x12\r\n\x05topic\x18\x02 \x01(\t\x12\x0e\n\x06offset\x18\x03 \x01(\x03\x12\x18\n\x0b\x63onsumer_id\x18\x04 \x01(\tH\x00\x88\x01\x01\x42\x0e\n\x0c_consumer_id\"M\n\x0cNackResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x15\n\rrouted_to_dlq\x18\x02 \x01(\x08\x12\x15\n\rerror_message\x18\x03 \x01(\t\"P\n\x0fMessageEnvelope\x12,\n\x04type\x18\x01 \x01(\x0e\x32\x1e.com.drmq.protocol.MessageType\x12\x0f\n\x07payload\x18\x02 \x01(\x0c\"\x8b\x02\n\tRaftEntry\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\r\n\x05index\x18\x02 \x01(\x03\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x0f\n\x07payload\x18\x04 \x01(\x0c\x12\x10\n\x03key\x18\x05 \x01(\tH\x00\x88\x01\x01\x12\x11\n\ttimestamp\x18\x06 \x01(\x03\x12\x38\n\x0c\x63ommand_type\x18\x07 \x01(\x0e\x32\".com.drmq.protocol.RaftCommandType\x12\x1b\n\x0e\x63onsumer_group\x18\x08 \x01(\tH\x01\x88\x01\x01\x12\x19\n\x0coffset_value\x18\t \x01(\x03H\x02\x88\x01\x01\x42\x06\n\x04_keyB\x11\n\x0f_consumer_groupB\x0f\n\r_offset_value\"g\n\x12RequestVoteRequest\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x14\n\x0c\x63\x61ndidate_id\x18\x02 \x01(\t\x12\x16\n\x0elast_log_index\x18\x03 \x01(\x03\x12\x15\n\rlast_log_term\x18\x04 \x01(\x03\"9\n\x13RequestVoteResponse\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x14\n\x0cvote_granted\x18\x02 \x01(\x08\"c\n\x0ePreVoteRequest\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x14\n\x0c\x63\x61ndidate_id\x18\x02 \x01(\t\x12\x16\n\x0elast_log_index\x18\x03 \x01(\x03\x12\x15\n\rlast_log_term\x18\x04 \x01(\x03\"5\n\x0fPreVoteResponse\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x14\n\x0cvote_granted\x18\x02 \x01(\x08\"\xac\x01\n\x14\x41ppendEntriesRequest\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x11\n\tleader_id\x18\x02 \x01(\t\x12\x16\n\x0eprev_log_index\x18\x03 \x01(\x03\x12\x15\n\rprev_log_term\x18\x04 \x01(\x03\x12-\n\x07\x65ntries\x18\x05 \x03(\x0b\x32\x1c.com.drmq.protocol.RaftEntry\x12\x15\n\rleader_commit\x18\x06 \x01(\x03\"K\n\x15\x41ppendEntriesResponse\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x0f\n\x07success\x18\x02 \x01(\x08\x12\x13\n\x0bmatch_index\x18\x03 \x01(\x03\"\x9e\x01\n\x16InstallSnapshotRequest\x12\x0c\n\x04term\x18\x01 \x01(\x03\x12\x11\n\tleader_id\x18\x02 \x01(\t\x12\x1b\n\x13last_included_index\x18\x03 \x01(\x03\x12\x1a\n\x12last_included_term\x18\x04 \x01(\x03\x12\x0e\n\x06offset\x18\x05 \x01(\x03\x12\x0c\n\x04\x64\x61ta\x18\x06 \x01(\x0c\x12\x0c\n\x04\x64one\x18\x07 \x01(\x08\"\'\n\x17InstallSnapshotResponse\x12\x0c\n\x04term\x18\x01 \x01(\x03*D\n\x0fRaftCommandType\x12\x0b\n\x07MESSAGE\x10\x00\x12\x11\n\rOFFSET_COMMIT\x10\x01\x12\x11\n\rBATCH_MESSAGE\x10\x02*\xe6\x04\n\x0bMessageType\x12\x0b\n\x07UNKNOWN\x10\x00\x12\x13\n\x0fPRODUCE_REQUEST\x10\x01\x12\x14\n\x10PRODUCE_RESPONSE\x10\x02\x12\x13\n\x0f\x43ONSUME_REQUEST\x10\x03\x12\x14\n\x10\x43ONSUME_RESPONSE\x10\x04\x12\r\n\tHEARTBEAT\x10\x05\x12\x19\n\x15\x43OMMIT_OFFSET_REQUEST\x10\x06\x12\x1a\n\x16\x43OMMIT_OFFSET_RESPONSE\x10\x07\x12\x18\n\x14\x46\x45TCH_OFFSET_REQUEST\x10\x08\x12\x19\n\x15\x46\x45TCH_OFFSET_RESPONSE\x10\t\x12\x18\n\x14REQUEST_VOTE_REQUEST\x10\n\x12\x19\n\x15REQUEST_VOTE_RESPONSE\x10\x0b\x12\x1a\n\x16\x41PPEND_ENTRIES_REQUEST\x10\x0c\x12\x1b\n\x17\x41PPEND_ENTRIES_RESPONSE\x10\r\x12\x14\n\x10PRE_VOTE_REQUEST\x10\x0e\x12\x15\n\x11PRE_VOTE_RESPONSE\x10\x0f\x12\x1c\n\x18INSTALL_SNAPSHOT_REQUEST\x10\x10\x12\x1d\n\x19INSTALL_SNAPSHOT_RESPONSE\x10\x11\x12\x10\n\x0cNACK_REQUEST\x10\x12\x12\x11\n\rNACK_RESPONSE\x10\x13\x12\x19\n\x15PRODUCE_BATCH_REQUEST\x10\x14\x12\x1a\n\x16PRODUCE_BATCH_RESPONSE\x10\x15\x12!\n\x1dSEARCH_OFFSET_BY_TIME_REQUEST\x10\x16\x12\"\n\x1eSEARCH_OFFSET_BY_TIME_RESPONSE\x10\x17*8\n\tErrorCode\x12\x08\n\x04NONE\x10\x00\x12\x0e\n\nNOT_LEADER\x10\x01\x12\x11\n\rUNKNOWN_ERROR\x10\x63\x42#\n\x11\x63om.drmq.protocolB\x0c\x44RMQProtocolP\x00\x62\x06proto3') - DESCRIPTOR._options = None - DESCRIPTOR._serialized_options = b'\n\021com.drmq.protocolB\014DRMQProtocolP\000' - _RAFTCOMMANDTYPE._serialized_start=2565 - _RAFTCOMMANDTYPE._serialized_end=2633 - _MESSAGETYPE._serialized_start=2636 - _MESSAGETYPE._serialized_end=3179 - _PRODUCEREQUEST._serialized_start=37 - _PRODUCEREQUEST._serialized_end=130 - _PRODUCERESPONSE._serialized_start=132 - _PRODUCERESPONSE._serialized_end=205 - _PRODUCEBATCHREQUEST._serialized_start=208 - _PRODUCEBATCHREQUEST._serialized_end=395 - _PRODUCEBATCHREQUEST_BATCHENTRY._serialized_start=314 - _PRODUCEBATCHREQUEST_BATCHENTRY._serialized_end=395 - _PRODUCEBATCHRESPONSE._serialized_start=397 - _PRODUCEBATCHRESPONSE._serialized_end=495 - _STOREDMESSAGE._serialized_start=497 - _STOREDMESSAGE._serialized_end=624 - _CONSUMEREQUEST._serialized_start=627 - _CONSUMEREQUEST._serialized_end=811 - _CONSUMERESPONSE._serialized_start=813 - _CONSUMERESPONSE._serialized_end=922 - _COMMITOFFSETREQUEST._serialized_start=924 - _COMMITOFFSETREQUEST._serialized_end=1042 - _COMMITOFFSETRESPONSE._serialized_start=1044 - _COMMITOFFSETRESPONSE._serialized_end=1106 - _FETCHOFFSETREQUEST._serialized_start=1108 - _FETCHOFFSETREQUEST._serialized_end=1167 - _FETCHOFFSETRESPONSE._serialized_start=1169 - _FETCHOFFSETRESPONSE._serialized_end=1246 - _NACKREQUEST._serialized_start=1248 - _NACKREQUEST._serialized_end=1358 - _NACKRESPONSE._serialized_start=1360 - _NACKRESPONSE._serialized_end=1437 - _MESSAGEENVELOPE._serialized_start=1439 - _MESSAGEENVELOPE._serialized_end=1519 - _RAFTENTRY._serialized_start=1522 - _RAFTENTRY._serialized_end=1789 - _REQUESTVOTEREQUEST._serialized_start=1791 - _REQUESTVOTEREQUEST._serialized_end=1894 - _REQUESTVOTERESPONSE._serialized_start=1896 - _REQUESTVOTERESPONSE._serialized_end=1953 - _PREVOTEREQUEST._serialized_start=1955 - _PREVOTEREQUEST._serialized_end=2054 - _PREVOTERESPONSE._serialized_start=2056 - _PREVOTERESPONSE._serialized_end=2109 - _APPENDENTRIESREQUEST._serialized_start=2112 - _APPENDENTRIESREQUEST._serialized_end=2284 - _APPENDENTRIESRESPONSE._serialized_start=2286 - _APPENDENTRIESRESPONSE._serialized_end=2361 - _INSTALLSNAPSHOTREQUEST._serialized_start=2364 - _INSTALLSNAPSHOTREQUEST._serialized_end=2522 - _INSTALLSNAPSHOTRESPONSE._serialized_start=2524 - _INSTALLSNAPSHOTRESPONSE._serialized_end=2563 +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'messages_pb2', _globals) +if not _descriptor._USE_C_DESCRIPTORS: + _globals['DESCRIPTOR']._loaded_options = None + _globals['DESCRIPTOR']._serialized_options = b'\n\021com.drmq.protocolB\014DRMQProtocolP\000' + _globals['_RAFTCOMMANDTYPE']._serialized_start=2815 + _globals['_RAFTCOMMANDTYPE']._serialized_end=2883 + _globals['_MESSAGETYPE']._serialized_start=2886 + _globals['_MESSAGETYPE']._serialized_end=3500 + _globals['_ERRORCODE']._serialized_start=3502 + _globals['_ERRORCODE']._serialized_end=3558 + _globals['_PRODUCEREQUEST']._serialized_start=37 + _globals['_PRODUCEREQUEST']._serialized_end=130 + _globals['_PRODUCERESPONSE']._serialized_start=132 + _globals['_PRODUCERESPONSE']._serialized_end=255 + _globals['_PRODUCEBATCHREQUEST']._serialized_start=258 + _globals['_PRODUCEBATCHREQUEST']._serialized_end=445 + _globals['_PRODUCEBATCHREQUEST_BATCHENTRY']._serialized_start=364 + _globals['_PRODUCEBATCHREQUEST_BATCHENTRY']._serialized_end=445 + _globals['_PRODUCEBATCHRESPONSE']._serialized_start=448 + _globals['_PRODUCEBATCHRESPONSE']._serialized_end=596 + _globals['_STOREDMESSAGE']._serialized_start=598 + _globals['_STOREDMESSAGE']._serialized_end=725 + _globals['_CONSUMEREQUEST']._serialized_start=728 + _globals['_CONSUMEREQUEST']._serialized_end=912 + _globals['_CONSUMERESPONSE']._serialized_start=914 + _globals['_CONSUMERESPONSE']._serialized_end=1023 + _globals['_COMMITOFFSETREQUEST']._serialized_start=1025 + _globals['_COMMITOFFSETREQUEST']._serialized_end=1143 + _globals['_COMMITOFFSETRESPONSE']._serialized_start=1145 + _globals['_COMMITOFFSETRESPONSE']._serialized_end=1207 + _globals['_FETCHOFFSETREQUEST']._serialized_start=1209 + _globals['_FETCHOFFSETREQUEST']._serialized_end=1268 + _globals['_FETCHOFFSETRESPONSE']._serialized_start=1270 + _globals['_FETCHOFFSETRESPONSE']._serialized_end=1347 + _globals['_SEARCHOFFSETBYTIMEREQUEST']._serialized_start=1349 + _globals['_SEARCHOFFSETBYTIMEREQUEST']._serialized_end=1410 + _globals['_SEARCHOFFSETBYTIMERESPONSE']._serialized_start=1412 + _globals['_SEARCHOFFSETBYTIMERESPONSE']._serialized_end=1496 + _globals['_NACKREQUEST']._serialized_start=1498 + _globals['_NACKREQUEST']._serialized_end=1608 + _globals['_NACKRESPONSE']._serialized_start=1610 + _globals['_NACKRESPONSE']._serialized_end=1687 + _globals['_MESSAGEENVELOPE']._serialized_start=1689 + _globals['_MESSAGEENVELOPE']._serialized_end=1769 + _globals['_RAFTENTRY']._serialized_start=1772 + _globals['_RAFTENTRY']._serialized_end=2039 + _globals['_REQUESTVOTEREQUEST']._serialized_start=2041 + _globals['_REQUESTVOTEREQUEST']._serialized_end=2144 + _globals['_REQUESTVOTERESPONSE']._serialized_start=2146 + _globals['_REQUESTVOTERESPONSE']._serialized_end=2203 + _globals['_PREVOTEREQUEST']._serialized_start=2205 + _globals['_PREVOTEREQUEST']._serialized_end=2304 + _globals['_PREVOTERESPONSE']._serialized_start=2306 + _globals['_PREVOTERESPONSE']._serialized_end=2359 + _globals['_APPENDENTRIESREQUEST']._serialized_start=2362 + _globals['_APPENDENTRIESREQUEST']._serialized_end=2534 + _globals['_APPENDENTRIESRESPONSE']._serialized_start=2536 + _globals['_APPENDENTRIESRESPONSE']._serialized_end=2611 + _globals['_INSTALLSNAPSHOTREQUEST']._serialized_start=2614 + _globals['_INSTALLSNAPSHOTREQUEST']._serialized_end=2772 + _globals['_INSTALLSNAPSHOTRESPONSE']._serialized_start=2774 + _globals['_INSTALLSNAPSHOTRESPONSE']._serialized_end=2813 # @@protoc_insertion_point(module_scope) diff --git a/drmq-ts-client/src/client.ts b/drmq-ts-client/src/client.ts index 9ad4242..34362fa 100644 --- a/drmq-ts-client/src/client.ts +++ b/drmq-ts-client/src/client.ts @@ -15,6 +15,8 @@ import { NackResponse, ProduceBatchRequest, ProduceBatchResponse, + SearchOffsetByTimeRequest, + SearchOffsetByTimeResponse, ErrorCode } from './messages'; @@ -349,6 +351,50 @@ export class DRMQConsumer extends DRMQClient { } } + public async seekByTime(topic: string, timestamp: number): Promise { + if (!this.subscriptions.includes(topic)) { + this.subscriptions.push(topic); + } + + for (let attempt = 0; attempt < this.maxRetries; attempt++) { + try { + await this.ensureConnected(); + + const req = SearchOffsetByTimeRequest.create({ + topic, + timestamp + }); + + const respBytes = await this.sendEnvelope( + MessageType.SEARCH_OFFSET_BY_TIME_REQUEST, + SearchOffsetByTimeRequest.encode(req).finish() + ); + + const resp = SearchOffsetByTimeResponse.decode(respBytes); + + if (resp.success) { + if (resp.offset >= 0) { + this.localOffsets[topic] = resp.offset; + if (this.groupMode) { + await this.commit(topic, resp.offset); + } + } + return; + } else if (await this.tryRedirectToLeader(resp.errorMessage)) { + continue; + } else { + throw new DRMQConnectionError(`Seek by time failed: ${resp.errorMessage}`); + } + } catch (err) { + if (err instanceof DRMQConnectionError && err.message.includes("Seek by time failed")) { + throw err; + } + await this.reconnect(); + } + } + throw new DRMQConnectionError(`Failed to seek offset by time after ${this.maxRetries} attempts`); + } + public async poll(maxMessages: number = 100, timeoutMs: number = 1000): Promise { for (let attempt = 0; attempt < this.maxRetries; attempt++) { try { diff --git a/drmq-ts-client/src/messages.ts b/drmq-ts-client/src/messages.ts index 090c3fa..47fabb0 100644 --- a/drmq-ts-client/src/messages.ts +++ b/drmq-ts-client/src/messages.ts @@ -75,6 +75,8 @@ export 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, UNRECOGNIZED = -1, } @@ -146,6 +148,12 @@ export function messageTypeFromJSON(object: any): MessageType { case 21: case "PRODUCE_BATCH_RESPONSE": return MessageType.PRODUCE_BATCH_RESPONSE; + case 22: + case "SEARCH_OFFSET_BY_TIME_REQUEST": + return MessageType.SEARCH_OFFSET_BY_TIME_REQUEST; + case 23: + case "SEARCH_OFFSET_BY_TIME_RESPONSE": + return MessageType.SEARCH_OFFSET_BY_TIME_RESPONSE; case -1: case "UNRECOGNIZED": default: @@ -199,6 +207,10 @@ export function messageTypeToJSON(object: MessageType): string { return "PRODUCE_BATCH_REQUEST"; case MessageType.PRODUCE_BATCH_RESPONSE: return "PRODUCE_BATCH_RESPONSE"; + case MessageType.SEARCH_OFFSET_BY_TIME_REQUEST: + return "SEARCH_OFFSET_BY_TIME_REQUEST"; + case MessageType.SEARCH_OFFSET_BY_TIME_RESPONSE: + return "SEARCH_OFFSET_BY_TIME_RESPONSE"; case MessageType.UNRECOGNIZED: default: return "UNRECOGNIZED"; @@ -361,6 +373,21 @@ export interface FetchOffsetResponse { errorMessage: string; } +/** Request to find an offset by a given timestamp */ +export interface SearchOffsetByTimeRequest { + topic: string; + /** The target timestamp */ + timestamp: number; +} + +/** Response with the found offset */ +export interface SearchOffsetByTimeResponse { + success: boolean; + /** The found offset, or -1 if none */ + offset: number; + errorMessage: string; +} + /** Request to negatively acknowledge (NACK) a message, triggering redelivery or DLQ routing */ export interface NackRequest { consumerGroup: string; @@ -1808,6 +1835,178 @@ export const FetchOffsetResponse: MessageFns = { }, }; +function createBaseSearchOffsetByTimeRequest(): SearchOffsetByTimeRequest { + return { topic: "", timestamp: 0 }; +} + +export const SearchOffsetByTimeRequest: MessageFns = { + encode(message: SearchOffsetByTimeRequest, writer: BinaryWriter = new BinaryWriter()): BinaryWriter { + if (message.topic !== "") { + writer.uint32(10).string(message.topic); + } + if (message.timestamp !== 0) { + writer.uint32(16).int64(message.timestamp); + } + return writer; + }, + + decode(input: BinaryReader | Uint8Array, length?: number): SearchOffsetByTimeRequest { + const reader = input instanceof BinaryReader ? input : new BinaryReader(input); + const end = length === undefined ? reader.len : reader.pos + length; + const message = createBaseSearchOffsetByTimeRequest(); + while (reader.pos < end) { + const tag = reader.uint32(); + switch (tag >>> 3) { + case 1: { + if (tag !== 10) { + break; + } + + message.topic = reader.string(); + continue; + } + case 2: { + if (tag !== 16) { + break; + } + + message.timestamp = longToNumber(reader.int64()); + continue; + } + } + if ((tag & 7) === 4 || tag === 0) { + break; + } + reader.skip(tag & 7); + } + return message; + }, + + fromJSON(object: any): SearchOffsetByTimeRequest { + return { + topic: isSet(object.topic) ? globalThis.String(object.topic) : "", + timestamp: isSet(object.timestamp) ? globalThis.Number(object.timestamp) : 0, + }; + }, + + toJSON(message: SearchOffsetByTimeRequest): unknown { + const obj: any = {}; + if (message.topic !== "") { + obj.topic = message.topic; + } + if (message.timestamp !== 0) { + obj.timestamp = Math.round(message.timestamp); + } + return obj; + }, + + create, I>>(base?: I): SearchOffsetByTimeRequest { + return SearchOffsetByTimeRequest.fromPartial(base ?? ({} as any)); + }, + fromPartial, I>>(object: I): SearchOffsetByTimeRequest { + const message = createBaseSearchOffsetByTimeRequest(); + message.topic = object.topic ?? ""; + message.timestamp = object.timestamp ?? 0; + return message; + }, +}; + +function createBaseSearchOffsetByTimeResponse(): SearchOffsetByTimeResponse { + return { success: false, offset: 0, errorMessage: "" }; +} + +export const SearchOffsetByTimeResponse: MessageFns = { + encode(message: SearchOffsetByTimeResponse, writer: BinaryWriter = new BinaryWriter()): BinaryWriter { + if (message.success !== false) { + writer.uint32(8).bool(message.success); + } + if (message.offset !== 0) { + writer.uint32(16).int64(message.offset); + } + if (message.errorMessage !== "") { + writer.uint32(26).string(message.errorMessage); + } + return writer; + }, + + decode(input: BinaryReader | Uint8Array, length?: number): SearchOffsetByTimeResponse { + const reader = input instanceof BinaryReader ? input : new BinaryReader(input); + const end = length === undefined ? reader.len : reader.pos + length; + const message = createBaseSearchOffsetByTimeResponse(); + while (reader.pos < end) { + const tag = reader.uint32(); + switch (tag >>> 3) { + case 1: { + if (tag !== 8) { + break; + } + + message.success = reader.bool(); + continue; + } + case 2: { + if (tag !== 16) { + break; + } + + message.offset = longToNumber(reader.int64()); + continue; + } + case 3: { + if (tag !== 26) { + break; + } + + message.errorMessage = reader.string(); + continue; + } + } + if ((tag & 7) === 4 || tag === 0) { + break; + } + reader.skip(tag & 7); + } + return message; + }, + + fromJSON(object: any): SearchOffsetByTimeResponse { + return { + success: isSet(object.success) ? globalThis.Boolean(object.success) : false, + offset: isSet(object.offset) ? globalThis.Number(object.offset) : 0, + errorMessage: isSet(object.errorMessage) + ? globalThis.String(object.errorMessage) + : isSet(object.error_message) + ? globalThis.String(object.error_message) + : "", + }; + }, + + toJSON(message: SearchOffsetByTimeResponse): unknown { + const obj: any = {}; + if (message.success !== false) { + obj.success = message.success; + } + if (message.offset !== 0) { + obj.offset = Math.round(message.offset); + } + if (message.errorMessage !== "") { + obj.errorMessage = message.errorMessage; + } + return obj; + }, + + create, I>>(base?: I): SearchOffsetByTimeResponse { + return SearchOffsetByTimeResponse.fromPartial(base ?? ({} as any)); + }, + fromPartial, I>>(object: I): SearchOffsetByTimeResponse { + const message = createBaseSearchOffsetByTimeResponse(); + message.success = object.success ?? false; + message.offset = object.offset ?? 0; + message.errorMessage = object.errorMessage ?? ""; + return message; + }, +}; + function createBaseNackRequest(): NackRequest { return { consumerGroup: "", topic: "", offset: 0, consumerId: undefined }; }