feat: add message timestamp seeking support to broker and update dashboard inspection UI

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

@@ -0,0 +1,206 @@
+package com.drmq.broker;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.util.List;
+
+/**
+ * Lightweight HTTP server for administrative REST APIs.
+ */
+public class AdminHttpServer {
+    private static final Logger logger = LoggerFactory.getLogger(AdminHttpServer.class);
+    private final HttpServer server;
+    private final MessageStore messageStore;
+    private final OffsetManager offsetManager;
+    private final ConsumerGroupCoordinator groupCoordinator;
+    private final Gson gson = new Gson();
+
+    public AdminHttpServer(int port, MessageStore messageStore, OffsetManager offsetManager, ConsumerGroupCoordinator groupCoordinator) throws IOException {
+        this.messageStore = messageStore;
+        this.offsetManager = offsetManager;
+        this.groupCoordinator = groupCoordinator;
+
+        this.server = HttpServer.create(new InetSocketAddress(port), 0);
+
+        this.server.createContext("/api/topics", this::handleTopics);
+        this.server.createContext("/api/consumers", this::handleConsumers);
+        this.server.createContext("/api/messages", this::handleMessages);
+
+        // CORS and standard executor
+        this.server.setExecutor(null); 
+    }
+
+    public void start() {
+        server.start();
+        logger.info("Admin HTTP Server started on port {}", server.getAddress().getPort());
+    }
+
+    public void stop() {
+        server.stop(1);
+        logger.info("Admin HTTP Server stopped.");
+    }
+
+    private void handleTopics(HttpExchange exchange) throws IOException {
+        addCorsHeaders(exchange);
+        if ("OPTIONS".equals(exchange.getRequestMethod())) {
+            exchange.sendResponseHeaders(204, -1);
+            return;
+        }
+
+        JsonArray topicsArray = new JsonArray();
+        List<String> topics = messageStore.getTopics();
+
+        for (String topic : topics) {
+            JsonObject obj = new JsonObject();
+            obj.addProperty("name", topic);
+            obj.addProperty("messageCount", messageStore.getMessageCount(topic));
+            // The global offset is global, not per topic. 
+            // We'll just return the message count for now.
+            topicsArray.add(obj);
+        }
+
+        sendJsonResponse(exchange, 200, gson.toJson(topicsArray));
+    }
+
+    private void handleConsumers(HttpExchange exchange) throws IOException {
+        addCorsHeaders(exchange);
+        if ("OPTIONS".equals(exchange.getRequestMethod())) {
+            exchange.sendResponseHeaders(204, -1);
+            return;
+        }
+
+        JsonArray groupsArray = new JsonArray();
+        java.util.Map<String, Long> allOffsets = offsetManager.getAllOffsets();
+
+        // Group by consumer group name
+        java.util.Map<String, JsonArray> groupsMap = new java.util.HashMap<>();
+
+        for (java.util.Map.Entry<String, Long> entry : allOffsets.entrySet()) {
+            String[] parts = entry.getKey().split("/");
+            if (parts.length != 2) continue;
+
+            String groupName = parts[0];
+            String topicName = parts[1];
+            long committedOffset = entry.getValue();
+
+            // Calculate lag using true topic head offset rather than message count
+            // Since DRMQ uses global offsets, messageCount does not correlate to the offset values.
+            long headOffset = messageStore.getHeadOffset(topicName);
+            long lag = 0;
+            if (headOffset >= 0) {
+                long effectiveCommitted = Math.max(0, committedOffset); // -1 means none committed
+                lag = Math.max(0, (headOffset + 1) - effectiveCommitted);
+            }
+
+            JsonObject topicObj = new JsonObject();
+            topicObj.addProperty("topic", topicName);
+            topicObj.addProperty("headOffset", headOffset);
+            topicObj.addProperty("committedOffset", committedOffset);
+            topicObj.addProperty("lag", lag);
+            topicObj.addProperty("activeMembers", groupCoordinator.getConsumerCount(groupName, topicName));
+
+            groupsMap.computeIfAbsent(groupName, k -> new JsonArray()).add(topicObj);
+        }
+
+        for (java.util.Map.Entry<String, JsonArray> entry : groupsMap.entrySet()) {
+            JsonObject groupObj = new JsonObject();
+            groupObj.addProperty("groupId", entry.getKey());
+            groupObj.add("topics", entry.getValue());
+            groupsArray.add(groupObj);
+        }
+
+        sendJsonResponse(exchange, 200, gson.toJson(groupsArray));
+    }
+
+    private void handleMessages(HttpExchange exchange) throws IOException {
+        addCorsHeaders(exchange);
+        if ("OPTIONS".equals(exchange.getRequestMethod())) {
+            exchange.sendResponseHeaders(204, -1);
+            return;
+        }
+
+        try {
+            String query = exchange.getRequestURI().getQuery();
+            if (query == null) {
+                sendJsonResponse(exchange, 400, "{\"error\":\"Missing query parameters\"}");
+                return;
+            }
+
+            String topic = null;
+            long offset = 0;
+            int limit = 10;
+            Long timestamp = null;
+
+            for (String param : query.split("&")) {
+                String[] pair = param.split("=");
+                if (pair.length == 2) {
+                    if ("topic".equals(pair[0])) topic = pair[1];
+                    else if ("offset".equals(pair[0])) offset = Long.parseLong(pair[1]);
+                    else if ("limit".equals(pair[0])) limit = Integer.parseInt(pair[1]);
+                    else if ("timestamp".equals(pair[0])) timestamp = Long.parseLong(pair[1]);
+                }
+            }
+
+            if (topic == null) {
+                sendJsonResponse(exchange, 400, "{\"error\":\"Missing 'topic' parameter\"}");
+                return;
+            }
+
+            // Limit bounds to avoid OOM
+            limit = Math.min(100, Math.max(1, limit));
+
+            if (timestamp != null && timestamp > 0) {
+                offset = messageStore.findOffsetByTimestamp(topic, timestamp);
+                if (offset == -1) {
+                    sendJsonResponse(exchange, 200, "[]");
+                    return;
+                }
+            }
+
+            List<com.drmq.protocol.DRMQProtocol.StoredMessage> messages = messageStore.getMessages(topic, offset, limit);
+            JsonArray msgsArray = new JsonArray();
+
+            for (com.drmq.protocol.DRMQProtocol.StoredMessage msg : messages) {
+                JsonObject obj = new JsonObject();
+                obj.addProperty("offset", msg.getOffset());
+                obj.addProperty("timestamp", msg.getTimestamp());
+                obj.addProperty("storedAt", msg.getStoredAt());
+                if (msg.hasKey()) {
+                    obj.addProperty("key", msg.getKey());
+                }
+                obj.addProperty("payload", msg.getPayload().toStringUtf8());
+                msgsArray.add(obj);
+            }
+
+            sendJsonResponse(exchange, 200, gson.toJson(msgsArray));
+        } catch (Exception e) {
+            logger.error("Error handling messages request", e);
+            sendJsonResponse(exchange, 500, "{\"error\":\"" + e.getMessage() + "\"}");
+        }
+    }
+
+    private void addCorsHeaders(HttpExchange exchange) {
+        exchange.getResponseHeaders().add("Access-Control-Allow-Origin", "*");
+        exchange.getResponseHeaders().add("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS");
+        exchange.getResponseHeaders().add("Access-Control-Allow-Headers", "Content-Type, Authorization");
+    }
+
+    private void sendJsonResponse(HttpExchange exchange, int statusCode, String response) throws IOException {
+        byte[] bytes = response.getBytes("UTF-8");
+        exchange.getResponseHeaders().add("Content-Type", "application/json");
+        exchange.sendResponseHeaders(statusCode, bytes.length);
+        try (OutputStream os = exchange.getResponseBody()) {
+            os.write(bytes);
+        }
+    }
+}

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

@@ -46,6 +46,7 @@ public class BrokerServer {
     private final List<RaftPeer> raftPeers; 
     private final BrokerMetrics metrics;
     private TelemetryWebSocketServer telemetryServer;
+    private AdminHttpServer adminHttpServer;
 
     private volatile boolean running = false;
 
@@ -171,6 +172,10 @@ public void initChannel(SocketChannel ch) {
             telemetryServer = new TelemetryWebSocketServer(wsPort, this);
             telemetryServer.start();
 
+            int adminPort = config.getPort() + 300;
+            adminHttpServer = new AdminHttpServer(adminPort, messageStore, offsetManager, groupCoordinator);
+            adminHttpServer.start();
+
             logger.info("DRMQ Broker started on port {} with data directory {}",
                     config.getPort(), config.getDataDir());
 
@@ -214,6 +219,9 @@ public void shutdown() {
         if (telemetryServer != null) {
             telemetryServer.shutdown();
         }
+        if (adminHttpServer != null) {
+            adminHttpServer.stop();
+        }
 
         if (activeChannels != null) {
             activeChannels.close().awaitUninterruptibly();

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

@@ -293,7 +293,6 @@ public int getActiveLeasesCount() {
         return total;
     }
 
-    // ---- Dead-Letter Queue (DLQ) Support ----
 
     /**
      * Explicitly reject (NACK) a message offset for a consumer within a group.
@@ -319,7 +318,6 @@ public boolean nackOffset(String group, String topic, String consumerId, long of
 
         state.lock.lock();
         try {
-            // Remove the consumer's active lease
             Lease lease = state.activeLeases.remove(consumerId);
             if (lease != null) {
                 state.members.remove(consumerId);
@@ -333,7 +331,6 @@ public boolean nackOffset(String group, String topic, String consumerId, long of
                 routeToDlq(state, group, topic, offset);
                 return true;
             } else {
-                // Rewind for redelivery using the lease's fromOffset to not skip messages
                 long rewindOffset = (lease != null) ? lease.fromOffset : offset;
                 if (rewindOffset < state.dispatchOffset) {
                     state.dispatchOffset = rewindOffset;
@@ -379,15 +376,13 @@ private void routeToDlq(GroupTopicState state, String group, String topic, long
             }
         });
 
-        // Advance past the bad offset regardless — don't let a DLQ write failure block progress
         long nextOffset = badOffset + 1;
         state.committedRanges.add(new CommittedRange(badOffset, nextOffset));
         advanceCommittedOffset(state, group, topic);
         if (state.dispatchOffset <= badOffset) {
             state.dispatchOffset = nextOffset;
         }
 
-        // Clean up the delivery counter for this offset
         state.deliveryCounts.remove(badOffset);
     }
 

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

@@ -48,6 +48,9 @@ public class MessageStore implements Closeable {
 
     // Topic -> Total number of messages
     private final ConcurrentHashMap<String, AtomicLong> topicMessageCounts = new ConcurrentHashMap<>();
+
+    // Topic -> Highest offset appended
+    private final ConcurrentHashMap<String, AtomicLong> topicHeadOffsets = new ConcurrentHashMap<>();
 
     // In-memory cache for recent messages (Topic -> BoundedMessageCache)
     private final ConcurrentHashMap<String, BoundedMessageCache> messageCache = new ConcurrentHashMap<>();
@@ -124,6 +127,11 @@ private void recoverInternal() throws IOException {
                         addToCache(topic, message);
                         topicMessageCounts.computeIfAbsent(topic, k -> new AtomicLong()).incrementAndGet();
 
+                        AtomicLong head = topicHeadOffsets.computeIfAbsent(topic, k -> new AtomicLong(-1));
+                        if (offset > head.get()) {
+                            head.set(offset);
+                        }
+
                         if (offset > maxOffset) {
                             maxOffset = offset;
                         }
@@ -156,6 +164,7 @@ public void reload() throws IOException {
             logger.info("Reloading MessageStore state from disk...");
             topicIndex.clear();
             topicMessageCounts.clear();
+            topicHeadOffsets.clear();
             messageCache.clear();
             topicWriteLocks.clear();
 
@@ -222,6 +231,11 @@ public long append(String topic, byte[] payload, String key, long clientTimestam
 
             addToCache(topic, message);
             topicMessageCounts.computeIfAbsent(topic, k -> new AtomicLong()).incrementAndGet();
+
+            AtomicLong head = topicHeadOffsets.computeIfAbsent(topic, k -> new AtomicLong(-1));
+            if (offset > head.get()) {
+                head.set(offset);
+            }
 
             logger.debug("Persisted and indexed message: topic={}, offset={}, position={}, segment={}", 
                     topic, offset, position, segment.getFilePath().getFileName());
@@ -292,11 +306,15 @@ public long appendBatch(String topic, List<ProduceBatchRequest.BatchEntry> entri
             }
 
             AtomicLong counter = topicMessageCounts.computeIfAbsent(topic, k -> new AtomicLong());
+            AtomicLong head = topicHeadOffsets.computeIfAbsent(topic, k -> new AtomicLong(-1));
             for (int i = 0; i < messages.size(); i++) {
                 StoredMessage msg = messages.get(i);
                 indexMessage(topic, msg.getOffset(), positions.get(i));
                 addToCache(topic, msg);
                 counter.incrementAndGet();
+                if (msg.getOffset() > head.get()) {
+                    head.set(msg.getOffset());
+                }
             }
 
             logger.debug("Batch persisted: topic={}, baseOffset={}, count={}", topic, baseOffset, batchSize);
@@ -523,6 +541,11 @@ public int getMessageCount(String topic) {
         return count == null ? 0 : count.intValue();
     }
 
+    public long getHeadOffset(String topic) {
+        AtomicLong head = topicHeadOffsets.get(topic);
+        return head == null ? -1 : head.get();
+    }
+
     public List<String> getTopics() {
         return new ArrayList<>(topicMessageCounts.keySet());
     }
@@ -542,6 +565,7 @@ public long getCachedMessageCount() {
     public void clear() {
         topicIndex.clear();
         topicMessageCounts.clear();
+        topicHeadOffsets.clear();
         messageCache.clear();
         globalOffset.set(0);
         logger.info("Message store memory state cleared");
@@ -682,9 +706,16 @@ public void removeRange(long fromOffset, long toOffset) {
         public List<StoredMessage> getMessagesFrom(long fromOffset, int maxCount) {
             lock.readLock().lock();
             try {
+                if (!cache.containsKey(fromOffset)) {
+                    return Collections.emptyList();
+                }
                 List<StoredMessage> result = new ArrayList<>();
+                boolean found = false;
                 for (Map.Entry<Long, StoredMessage> entry : cache.entrySet()) {
-                    if (entry.getKey() >= fromOffset) {
+                    if (entry.getKey() == fromOffset) {
+                        found = true;
+                    }
+                    if (found) {
                         result.add(entry.getValue());
                         if (result.size() >= maxCount) {
                             break;

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

@@ -74,6 +74,14 @@ public int getOffsetEntryCount() {
         return offsets.size();
     }
 
+    /**
+     * Get a snapshot of all committed offsets across all groups and topics.
+     * @return map of "group/topic" to offset
+     */
+    public Map<String, Long> getAllOffsets() {
+        return new java.util.HashMap<>(offsets);
+    }
+
     /** Load all offsets from disk on startup. */
     private void load() throws IOException {
         if (!Files.exists(offsetsFile)) {