feat: add message timestamp seeking support to broker and update dashboard inspection UI#34
feat: add message timestamp seeking support to broker and update dashboard inspection UI#34samuel025 wants to merge 3 commits into
Conversation
…hroughput formatting, and implement broker admin HTTP server.
…figuration and stress testing scripts
…board inspection UI
📝 WalkthroughWalkthroughChangesAdmin observability
Telemetry history and dashboard controls
Raft state visibility
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Dashboard
participant AdminHttpServer
participant MessageStore
participant OffsetManager
Dashboard->>AdminHttpServer: Request topics, consumers, or messages
AdminHttpServer->>MessageStore: Read topics, head offsets, or messages
AdminHttpServer->>OffsetManager: Read committed offsets
AdminHttpServer-->>Dashboard: Return JSON response
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 12
🧹 Nitpick comments (1)
drmq-dashboard/src/pages/Topics.tsx (1)
72-86: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdmin API base URL (
http://localhost:9392) is hardcoded and duplicated across pages. All three sites bake in the same non-configurable host/port, which only ever targets a single broker and will drift if the admin port scheme changes (BrokerServerderives it asconfig.getPort() + 300).
drmq-dashboard/src/pages/Topics.tsx#L72-L86: extract the base URL (fetchTopics) into a shared constant/env var (e.g.VITE_ADMIN_API_URL).drmq-dashboard/src/pages/Topics.tsx#L25-L60: use the same shared base URL constant infetchMessages's URL construction.drmq-dashboard/src/pages/Consumers.tsx#L23-L36: use the same shared base URL constant infetchConsumers.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@drmq-dashboard/src/pages/Topics.tsx` around lines 72 - 86, Replace the hardcoded admin API host in fetchTopics with a shared configurable base URL, preferably sourced from VITE_ADMIN_API_URL with the existing local value as its default. Reuse that same constant when constructing fetchMessages URLs in drmq-dashboard/src/pages/Topics.tsx lines 25-60 and fetchConsumers URLs in drmq-dashboard/src/pages/Consumers.tsx lines 23-36; all three sites require this change.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.java`:
- Around line 88-90: Update the key parsing in the allOffsets iteration to split
the entry key at only the first "/" so topics may contain additional slashes.
Keep the existing two-part validation and consumer-lag handling, while
preserving the known group-name ambiguity in the "group/topic" key format.
- Around line 186-189: Update the exception response in AdminHttpServer’s
message-request handler to construct the error object through the existing JSON
serializer (such as Gson) instead of concatenating e.getMessage() into a JSON
string. Ensure messages containing quotes or other special characters remain
valid JSON so Topics.tsx can read data.error.
- Around line 20-41: Update the AdminHttpServer constructor and request-handling
flow to bind HttpServer to loopback instead of all interfaces by default, and
require authentication before serving /api/topics, /api/consumers, or
/api/messages. Replace the wildcard CORS behavior with an authenticated,
restricted origin policy, using the existing handlers such as handleTopics,
handleConsumers, and handleMessages without exposing data to unauthenticated
clients.
- Line 40: Replace the null executor configured in AdminHttpServer with a small
dedicated executor for the admin HTTP server, using the server’s setExecutor
call. Ensure the pool is suitable for handling concurrent dashboard endpoints so
slow /api/messages requests do not block /api/topics or /api/consumers.
In `@drmq-broker/src/main/java/com/drmq/broker/MessageStore.java`:
- Around line 234-238: Make the update to topicHeadOffsets in the append path
atomic: replace the separate head.get()/head.set() comparison with an AtomicLong
update operation that retains the maximum of the current value and offset. Keep
the existing computeIfAbsent initialization and topic-scoped behavior unchanged.
- Around line 709-718: Update the cache lookup in getMessagesFrom to avoid
requiring an exact fromOffset key: iterate cached entries and collect those
whose keys are greater than or equal to fromOffset. Remove or adjust the
containsKey(fromOffset) early return so normal consumption can use the bounded
cache fast path, while preserving the existing result construction and ordering
behavior.
- Around line 309-317: Update the head-offset update inside the batch loop in
MessageStore so the comparison and update use an atomic max operation on the
per-topic AtomicLong, matching the concurrency-safe behavior required by
append(). Preserve the existing indexing, caching, and counter updates.
In `@drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java`:
- Around line 48-57: Add a lock-protected immutable Raft status snapshot
containing state, currentTerm, leaderId, commitIndex, and lastApplied, with a
single accessor that captures all values atomically. Update
buildTelemetryPayload() to use this snapshot instead of separately calling the
individual getters, while preserving the existing telemetry fields and values.
In `@drmq-dashboard/src/pages/Dashboard.tsx`:
- Around line 333-340: Add an accessible name to the range input controlling
historyWindowSeconds, using either a visible label associated with the input or
aria-label="Throughput history window". Preserve the existing slider behavior
and styling.
- Around line 331-346: Update MockTelemetryProvider and its mock-history
initialization to retain and seed 300 samples, matching the maximum
historyWindowSeconds selectable by the Dashboard range control. Ensure
non-WebSocket charts display the full selected 60–300 second window while
preserving the existing 30-second sampling behavior.
In `@drmq-dashboard/src/pages/Topics.tsx`:
- Around line 25-70: Update the inspector-opening flow around fetchMessages and
the inspectorTopic useEffect so the initial fetch explicitly uses offset 0
rather than the stale inspectOffset captured before setInspectOffset completes.
Reset inspectTimestamp when opening a new topic, and ensure the first request
does not apply the previous timestamp filter; preserve existing offset/timestamp
behavior for subsequent fetches.
In `@drmq-python-client/test_time_seek.py`:
- Line 9: Update the test topic construction in test_time_seek.py to append a
UUID-based suffix instead of relying on int(time.time()) alone. Preserve the
existing topic prefix while ensuring each test invocation receives a
collision-proof unique topic.
---
Nitpick comments:
In `@drmq-dashboard/src/pages/Topics.tsx`:
- Around line 72-86: Replace the hardcoded admin API host in fetchTopics with a
shared configurable base URL, preferably sourced from VITE_ADMIN_API_URL with
the existing local value as its default. Reuse that same constant when
constructing fetchMessages URLs in drmq-dashboard/src/pages/Topics.tsx lines
25-60 and fetchConsumers URLs in drmq-dashboard/src/pages/Consumers.tsx lines
23-36; all three sites require this change.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: d7fe8a45-1a10-4af3-841e-967e15f2ccf4
📒 Files selected for processing (12)
drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.javadrmq-broker/src/main/java/com/drmq/broker/BrokerServer.javadrmq-broker/src/main/java/com/drmq/broker/ConsumerGroupCoordinator.javadrmq-broker/src/main/java/com/drmq/broker/MessageStore.javadrmq-broker/src/main/java/com/drmq/broker/OffsetManager.javadrmq-broker/src/main/java/com/drmq/broker/TelemetryWebSocketServer.javadrmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.javadrmq-dashboard/src/App.tsxdrmq-dashboard/src/pages/Consumers.tsxdrmq-dashboard/src/pages/Dashboard.tsxdrmq-dashboard/src/pages/Topics.tsxdrmq-python-client/test_time_seek.py
💤 Files with no reviewable changes (1)
- drmq-broker/src/main/java/com/drmq/broker/ConsumerGroupCoordinator.java
| 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); | ||
| } |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -n --type=java -C2 'Authorization|BasicAuth|ApiKey|apiKey|Authenticat' drmq-broker/src/main/java/com/drmq/brokerRepository: samuel025/DRMQ
Length of output: 784
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- AdminHttpServer.java outline ---'
ast-grep outline drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.java --view expanded || true
echo '--- Relevant AdminHttpServer.java slices ---'
sed -n '1,260p' drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.java
echo '--- Search for admin server construction / port binding / auth config ---'
rg -n --type=java -C2 'new AdminHttpServer|admin.*port|adminPort|HttpServer.create|setExecutor\(|Access-Control-Allow-Origin|Authorization|Basic|auth' drmq-broker/src/main/javaRepository: samuel025/DRMQ
Length of output: 12774
Lock down the admin API
HttpServer.create(new InetSocketAddress(port), 0) binds to all interfaces, and the handlers also return Access-Control-Allow-Origin: * with no auth check. That makes /api/messages and /api/consumers readable by any reachable client, including raw payloads and consumer offsets. Bind to loopback by default and add authentication before exposing this outside local development.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.java` around lines
20 - 41, Update the AdminHttpServer constructor and request-handling flow to
bind HttpServer to loopback instead of all interfaces by default, and require
authentication before serving /api/topics, /api/consumers, or /api/messages.
Replace the wildcard CORS behavior with an authenticated, restricted origin
policy, using the existing handlers such as handleTopics, handleConsumers, and
handleMessages without exposing data to unauthenticated clients.
| this.server.createContext("/api/messages", this::handleMessages); | ||
|
|
||
| // CORS and standard executor | ||
| this.server.setExecutor(null); |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
sed -n '1,220p' drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.javaRepository: samuel025/DRMQ
Length of output: 8486
🌐 Web query:
JDK HttpServer.setExecutor(null) default executor single threaded behavior
💡 Result:
When you call HttpServer.setExecutor(null) (or fail to set an executor before starting the server), the JDK HttpServer uses a default executor implementation [1][2][3]. This default executor is synchronous and single-threaded [4][5]. Internally, the server uses a static class called DefaultExecutor, which executes tasks by calling task.run directly within the thread that manages the request dispatching [4][5]. Consequently, all HTTP requests are handled serially in the same thread, meaning they are processed in a blocking manner [5]. As a result, the server cannot process multiple requests in parallel when using the default executor [5]. To enable parallel request processing, you must provide a custom Executor, such as one returned by java.util.concurrent.Executors (e.g., Executors.newCachedThreadPool or Executors.newFixedThreadPool) [5].
Citations:
- 1: https://docs.oracle.com/en/java/javase/25/docs/api/jdk.httpserver/com/sun/net/httpserver/HttpServer.html
- 2: https://docs.oracle.com/en/java/javase/17/docs/api/jdk.httpserver/com/sun/net/httpserver/HttpServer.html
- 3: https://docs.oracle.com/en/java/javase/11/docs/api/jdk.httpserver/com/sun/net/httpserver/HttpServer.html
- 4: https://github.com/openjdk/jdk/blob/master/src/jdk.httpserver/share/classes/sun/net/httpserver/ServerImpl.java
- 5: https://stackoverflow.com/questions/14729475/can-i-make-a-java-httpserver-threaded-process-requests-in-parallel
Use a dedicated executor for the admin server. setExecutor(null) leaves requests on the JDK default executor, so one slow /api/messages call can block /api/topics and /api/consumers. Use a small dedicated pool here to keep the dashboard responsive.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.java` at line 40,
Replace the null executor configured in AdminHttpServer with a small dedicated
executor for the admin HTTP server, using the server’s setExecutor call. Ensure
the pool is suitable for handling concurrent dashboard endpoints so slow
/api/messages requests do not block /api/topics or /api/consumers.
| for (java.util.Map.Entry<String, Long> entry : allOffsets.entrySet()) { | ||
| String[] parts = entry.getKey().split("/"); | ||
| if (parts.length != 2) continue; |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
split("/") silently drops any consumer/topic entry whose topic name contains "/".
OffsetManager.key() joins as group + "/" + topic. If topic itself contains a "/", split("/") yields more than 2 parts and the if (parts.length != 2) continue; guard silently skips that group/topic pair from the consumer-lag view instead of parsing it correctly.
🐛 Proposed fix: limit split to 2 parts
- String[] parts = entry.getKey().split("/");
+ String[] parts = entry.getKey().split("/", 2);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| for (java.util.Map.Entry<String, Long> entry : allOffsets.entrySet()) { | |
| String[] parts = entry.getKey().split("/"); | |
| if (parts.length != 2) continue; | |
| for (java.util.Map.Entry<String, Long> entry : allOffsets.entrySet()) { | |
| String[] parts = entry.getKey().split("/", 2); | |
| if (parts.length != 2) continue; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.java` around lines
88 - 90, Update the key parsing in the allOffsets iteration to split the entry
key at only the first "/" so topics may contain additional slashes. Keep the
existing two-part validation and consumer-lag handling, while preserving the
known group-name ambiguity in the "group/topic" key format.
| } catch (Exception e) { | ||
| logger.error("Error handling messages request", e); | ||
| sendJsonResponse(exchange, 500, "{\"error\":\"" + e.getMessage() + "\"}"); | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Unescaped exception message can produce malformed JSON error responses.
e.getMessage() is concatenated directly into a JSON string literal. NumberFormatException messages (a likely cause here, from Long.parseLong/Integer.parseInt on bad query params) contain embedded double quotes (e.g. For input string: "abc"), which breaks the resulting JSON and defeats Topics.tsx's data.error handling.
🐛 Proposed fix using Gson to build the error object
} catch (Exception e) {
logger.error("Error handling messages request", e);
- sendJsonResponse(exchange, 500, "{\"error\":\"" + e.getMessage() + "\"}");
+ JsonObject err = new JsonObject();
+ err.addProperty("error", e.getMessage());
+ sendJsonResponse(exchange, 500, gson.toJson(err));
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| } catch (Exception e) { | |
| logger.error("Error handling messages request", e); | |
| sendJsonResponse(exchange, 500, "{\"error\":\"" + e.getMessage() + "\"}"); | |
| } | |
| } catch (Exception e) { | |
| logger.error("Error handling messages request", e); | |
| JsonObject err = new JsonObject(); | |
| err.addProperty("error", e.getMessage()); | |
| sendJsonResponse(exchange, 500, gson.toJson(err)); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.java` around lines
186 - 189, Update the exception response in AdminHttpServer’s message-request
handler to construct the error object through the existing JSON serializer (such
as Gson) instead of concatenating e.getMessage() into a JSON string. Ensure
messages containing quotes or other special characters remain valid JSON so
Topics.tsx can read data.error.
|
|
||
| AtomicLong head = topicHeadOffsets.computeIfAbsent(topic, k -> new AtomicLong(-1)); | ||
| if (offset > head.get()) { | ||
| head.set(offset); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Non-atomic "set-if-greater" on topicHeadOffsets races under concurrent appends to the same topic.
head.get()/head.set() here runs outside the synchronized(topicLock) block, only under the shared globalLock.readLock(). Two threads appending to the same topic concurrently can interleave the check-then-set and leave head at a stale, lower value than the true max offset — this directly feeds getHeadOffset(), which AdminHttpServer.handleConsumers uses to compute consumer lag.
🔒 Proposed fix using atomic max update
- AtomicLong head = topicHeadOffsets.computeIfAbsent(topic, k -> new AtomicLong(-1));
- if (offset > head.get()) {
- head.set(offset);
- }
+ topicHeadOffsets.computeIfAbsent(topic, k -> new AtomicLong(-1))
+ .accumulateAndGet(offset, Math::max);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| AtomicLong head = topicHeadOffsets.computeIfAbsent(topic, k -> new AtomicLong(-1)); | |
| if (offset > head.get()) { | |
| head.set(offset); | |
| } | |
| topicHeadOffsets.computeIfAbsent(topic, k -> new AtomicLong(-1)) | |
| .accumulateAndGet(offset, Math::max); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@drmq-broker/src/main/java/com/drmq/broker/MessageStore.java` around lines 234
- 238, Make the update to topicHeadOffsets in the append path atomic: replace
the separate head.get()/head.set() comparison with an AtomicLong update
operation that retains the maximum of the current value and offset. Keep the
existing computeIfAbsent initialization and topic-scoped behavior unchanged.
| private volatile long currentTerm; | ||
| private volatile String votedFor; | ||
| private final RaftLog raftLog; | ||
|
|
||
| // Volatile state | ||
| private RaftState state; | ||
| private long commitIndex; | ||
| private long lastApplied; | ||
| private long lastAppliedTerm; | ||
| private String leaderId; | ||
| private volatile RaftState state; | ||
| private volatile long commitIndex; | ||
| private volatile long lastApplied; | ||
| private volatile long lastAppliedTerm; | ||
| private volatile String leaderId; |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the relevant files and their structure around the referenced lines.
git ls-files | rg 'drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java|TelemetryWebSocketServer.java$'
echo '--- RaftNode outline ---'
ast-grep outline drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java --view expanded || true
echo '--- TelemetryWebSocketServer outline ---'
ast-grep outline drmq-broker/src/main/java/com/drmq/broker/raft/TelemetryWebSocketServer.java --view expanded || true
echo '--- RaftNode excerpt ---'
sed -n '1,260p' drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java
echo '--- TelemetryWebSocketServer excerpt ---'
sed -n '200,320p' drmq-broker/src/main/java/com/drmq/broker/raft/TelemetryWebSocketServer.javaRepository: samuel025/DRMQ
Length of output: 14114
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Focus on the likely getters and mutation sites if the excerpts are large.
rg -n "getCurrentTerm|getVotedFor|getState|getCommitIndex|getLastApplied|getLastAppliedTerm|getLeaderId|stepDown|apply|leaderId|commitIndex|lastApplied" \
drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java \
drmq-broker/src/main/java/com/drmq/broker/raft/TelemetryWebSocketServer.javaRepository: samuel025/DRMQ
Length of output: 12087
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check whether the relevant fields are read together anywhere else.
rg -n "currentTerm|votedFor|state|commitIndex|lastApplied|lastAppliedTerm|leaderId" drmq-broker/src/main/java/com/drmq/broker/raft -g '!**/target/**'Repository: samuel025/DRMQ
Length of output: 23585
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Read only the minimal relevant sections.
for f in \
drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java \
drmq-broker/src/main/java/com/drmq/broker/raft/TelemetryWebSocketServer.java
do
echo "===== $f ====="
wc -l "$f"
done
echo '===== RaftNode relevant section ====='
sed -n '1,180p' drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java
echo '===== TelemetryWebSocketServer relevant section ====='
sed -n '220,300p' drmq-broker/src/main/java/com/drmq/broker/raft/TelemetryWebSocketServer.javaRepository: samuel025/DRMQ
Length of output: 488
🏁 Script executed:
#!/bin/bash
set -euo pipefail
f=drmq-broker/src/main/java/com/drmq/broker/TelemetryWebSocketServer.java
echo '--- outline ---'
ast-grep outline "$f" --view expanded || true
echo '--- file size ---'
wc -l "$f"
echo '--- relevant regions ---'
# Read likely telemetry/status methods without dumping the whole file if large.
rg -n "getState\\(|getCurrentTerm\\(|getLeaderId\\(|getCommitIndex\\(|getLastApplied\\(|status|telemetry|publish|websocket|snapshot" "$f"
echo '--- excerpt 1 ---'
sed -n '1,260p' "$f"
echo '--- excerpt 2 ---'
sed -n '260,420p' "$f"Repository: samuel025/DRMQ
Length of output: 21763
Expose a coherent Raft status snapshot.
buildTelemetryPayload() reads getState(), getCurrentTerm(), getLeaderId(), getCommitIndex(), and getLastApplied() separately, so telemetry can publish a mixed Raft state during role changes or snapshot apply. Add a lock-protected immutable snapshot and use it here.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java` around lines 48
- 57, Add a lock-protected immutable Raft status snapshot containing state,
currentTerm, leaderId, commitIndex, and lastApplied, with a single accessor that
captures all values atomically. Update buildTelemetryPayload() to use this
snapshot instead of separately calling the individual getters, while preserving
the existing telemetry fields and values.
| <div className="mt-4 pt-4 border-t border-white/5 flex items-center justify-between gap-4"> | ||
| <span className="mono text-[10px] text-zinc-500 w-16">30s</span> | ||
| <input | ||
| type="range" | ||
| min="30" | ||
| max="300" | ||
| step="30" | ||
| value={historyWindowSeconds} | ||
| onChange={(e) => setHistoryWindowSeconds(Number(e.target.value))} | ||
| className="flex-1 h-1.5 bg-white/10 rounded-full appearance-none [&::-webkit-slider-thumb]:appearance-none [&::-webkit-slider-thumb]:w-3 [&::-webkit-slider-thumb]:h-3 [&::-webkit-slider-thumb]:bg-cyan-400 [&::-webkit-slider-thumb]:rounded-full cursor-pointer" | ||
| /> | ||
| <span className="mono text-[10px] text-zinc-500 w-16 text-right">5min</span> | ||
| </div> | ||
| <div className="text-center mt-1"> | ||
| <span className="mono text-[9px] text-zinc-600 tracking-wider">VIEWING LAST {historyWindowSeconds >= 60 ? `${historyWindowSeconds / 60} MIN` : `${historyWindowSeconds} SEC`}</span> | ||
| </div> |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Keep mock history length consistent with the selectable window.
When VITE_USE_WEBSOCKET is not true, the dashboard uses MockTelemetryProvider, which retains only 30 samples. Selecting 60–300 seconds therefore leaves the chart at 30 seconds while claiming a longer window. Seed and retain 300 mock samples, or cap the control to available history.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@drmq-dashboard/src/pages/Dashboard.tsx` around lines 331 - 346, Update
MockTelemetryProvider and its mock-history initialization to retain and seed 300
samples, matching the maximum historyWindowSeconds selectable by the Dashboard
range control. Ensure non-WebSocket charts display the full selected 60–300
second window while preserving the existing 30-second sampling behavior.
| <input | ||
| type="range" | ||
| min="30" | ||
| max="300" | ||
| step="30" | ||
| value={historyWindowSeconds} | ||
| onChange={(e) => setHistoryWindowSeconds(Number(e.target.value))} | ||
| className="flex-1 h-1.5 bg-white/10 rounded-full appearance-none [&::-webkit-slider-thumb]:appearance-none [&::-webkit-slider-thumb]:w-3 [&::-webkit-slider-thumb]:h-3 [&::-webkit-slider-thumb]:bg-cyan-400 [&::-webkit-slider-thumb]:rounded-full cursor-pointer" |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Give the range control an accessible name.
Add a <label> or aria-label="Throughput history window" so assistive technology can identify this slider.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@drmq-dashboard/src/pages/Dashboard.tsx` around lines 333 - 340, Add an
accessible name to the range input controlling historyWindowSeconds, using
either a visible label associated with the input or aria-label="Throughput
history window". Preserve the existing slider behavior and styling.
| const fetchMessages = async () => { | ||
| if (!inspectorTopic) return; | ||
| setInspectLoading(true); | ||
| setInspectError(null); | ||
| try { | ||
| const offsetParam = inspectOffset === '' ? 0 : inspectOffset; | ||
| const limitParam = inspectLimit === '' ? 1 : inspectLimit; | ||
|
|
||
| let url = `http://localhost:9392/api/messages?topic=${inspectorTopic}&limit=${limitParam}`; | ||
| if (inspectTimestamp) { | ||
| const ts = new Date(inspectTimestamp).getTime(); | ||
| if (!isNaN(ts)) { | ||
| url += `×tamp=${ts}`; | ||
| } else { | ||
| url += `&offset=${offsetParam}`; | ||
| } | ||
| } else { | ||
| url += `&offset=${offsetParam}`; | ||
| } | ||
|
|
||
| const res = await fetch(url); | ||
| if (!res.ok) throw new Error(`HTTP error! status: ${res.status}`); | ||
| const data = await res.json(); | ||
| if (data.error) throw new Error(data.error); | ||
| setMessages(data); | ||
|
|
||
| // Snap the input offset to the actual returned offset to show compaction jumps | ||
| if (data && data.length > 0) { | ||
| setInspectOffset(data[0].offset); | ||
| } | ||
| } catch (err: any) { | ||
| setInspectError(err.message || 'Failed to fetch messages'); | ||
| } finally { | ||
| setInspectLoading(false); | ||
| } | ||
| }; | ||
|
|
||
| // Automatically fetch messages when inspector opens | ||
| useEffect(() => { | ||
| if (inspectorTopic) { | ||
| setInspectOffset(0); // Reset to 0 when opening new topic | ||
| fetchMessages(); | ||
| } else { | ||
| setMessages([]); | ||
| } | ||
| }, [inspectorTopic]); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Opening the inspector for a new topic can fetch with a stale offset and a leftover timestamp filter from the previous topic.
setInspectOffset(0) only schedules a state update; the immediately-following fetchMessages() call in the same effect still closes over the pre-update inspectOffset (React batches the update), so the reset to 0 doesn't actually take effect for this first fetch. Additionally, inspectTimestamp is never reset here, so a timestamp seek left over from a previously inspected topic is silently reapplied to the newly opened topic. Since if (inspectTimestamp) takes priority over offset in fetchMessages, this can produce wrong/empty results for the new topic.
🐛 Proposed fix: pass an explicit override offset and reset the timestamp filter
- const fetchMessages = async () => {
+ const fetchMessages = async (overrideOffset?: number) => {
if (!inspectorTopic) return;
setInspectLoading(true);
setInspectError(null);
try {
- const offsetParam = inspectOffset === '' ? 0 : inspectOffset;
+ const offsetParam = overrideOffset ?? (inspectOffset === '' ? 0 : inspectOffset);
const limitParam = inspectLimit === '' ? 1 : inspectLimit; useEffect(() => {
if (inspectorTopic) {
setInspectOffset(0); // Reset to 0 when opening new topic
- fetchMessages();
+ setInspectTimestamp('');
+ fetchMessages(0);
} else {
setMessages([]);
}
}, [inspectorTopic]);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@drmq-dashboard/src/pages/Topics.tsx` around lines 25 - 70, Update the
inspector-opening flow around fetchMessages and the inspectorTopic useEffect so
the initial fetch explicitly uses offset 0 rather than the stale inspectOffset
captured before setInspectOffset completes. Reset inspectTimestamp when opening
a new topic, and ensure the first request does not apply the previous timestamp
filter; preserve existing offset/timestamp behavior for subsequent fetches.
|
|
||
| def run_test(): | ||
| servers = "localhost:9092,localhost:9093,localhost:9094" | ||
| topic = "time-seek-topic-" + str(int(time.time())) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Make the test topic collision-proof.
Second-resolution names collide for parallel or rapid runs, allowing another invocation’s messages to cause a false pass or failure. Use a UUID suffix.
Proposed fix
+import uuid
import time
...
- topic = "time-seek-topic-" + str(int(time.time()))
+ topic = f"time-seek-topic-{uuid.uuid4().hex}"📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| topic = "time-seek-topic-" + str(int(time.time())) | |
| import uuid | |
| import time | |
| ... | |
| topic = f"time-seek-topic-{uuid.uuid4().hex}" |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@drmq-python-client/test_time_seek.py` at line 9, Update the test topic
construction in test_time_seek.py to append a UUID-based suffix instead of
relying on int(time.time()) alone. Preserve the existing topic prefix while
ensuring each test invocation receives a collision-proof unique topic.
1 participant
Summary by CodeRabbit
New Features
Improvements