Cross topic production#35
Conversation
…hroughput formatting, and implement broker admin HTTP server.
…figuration and stress testing scripts
…board inspection UI
…ol and update client/broker infrastructure accordingly.
📝 WalkthroughWalkthroughThis change adds atomic multi-topic production across protocol, broker, Raft, and client implementations; introduces broker administration APIs and dashboard pages; and updates storage retrieval, telemetry, WebSocket resilience, Raft lifecycle behavior, configuration, examples, and integration tests. ChangesAtomic multi-topic production
Administrative APIs and dashboard
Runtime, telemetry, and Raft maintenance
Estimated code review effort: 5 (Critical) | ~120 minutes 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: 11
🧹 Nitpick comments (6)
drmq-dashboard/src/pages/Topics.tsx (1)
77-77: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winHardcoded
http://localhost:9392admin origin across dashboard pages. Both pages target a single broker's admin port on localhost, so the dashboard only ever inspects Broker-1 and breaks when served from any non-localhost origin or when the leader is a different node. Extract the admin base URL into shared config (e.g. an env-driven constant or the telemetry provider's node info).
drmq-dashboard/src/pages/Topics.tsx#L77-L77: replace the literal infetchTopics(and the/api/messagesliteral at L33) with the shared base URL.drmq-dashboard/src/pages/Consumers.tsx#L27-L27: replace the literal infetchConsumerswith the same shared base URL.🤖 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` at line 77, Replace the hardcoded admin origin in Topics.tsx, including both fetchTopics and the /api/messages request, and Consumers.tsx in fetchConsumers with one shared, environment- or telemetry-driven admin base URL. Define or reuse the shared configuration so all three requests resolve against the intended broker rather than localhost:9392.drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.java (1)
144-152: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winQuery parsing is fragile: no URL-decoding and split on every
=.
param.split("=")discards values containing=, and none of the params are URL-decoded, so topic names with reserved characters resolve incorrectly. Malformed numeric values also throwNumberFormatException, which surfaces as a 500 rather than a 400. Consider limiting the split and decoding, and returning 400 on parse failure.♻️ Sketch
for (String param : query.split("&")) { - String[] pair = param.split("="); + String[] pair = param.split("=", 2); if (pair.length == 2) { - if ("topic".equals(pair[0])) topic = pair[1]; + String value = java.net.URLDecoder.decode(pair[1], java.nio.charset.StandardCharsets.UTF_8); + if ("topic".equals(pair[0])) topic = value; ... } }🤖 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 144 - 152, Update the query parsing in AdminHttpServer to split each parameter only at its first “=” and URL-decode both parameter names and values before matching topic, offset, limit, and timestamp. Catch malformed numeric values from Long.parseLong and Integer.parseInt, and return an HTTP 400 response instead of allowing NumberFormatException to surface as a 500.drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java (1)
1551-1557: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueOptional: magic
-100retention margin in startup compaction.
compactUpTo = lastApplied - 100hardcodes a 100-entry safety margin. Consider extracting a named constant (mirroringraftCompactThresholdusage elsewhere) so the startup and runtime compaction margins are documented and tunable together.🤖 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 1551 - 1557, The startup compaction logic in RaftNode currently hardcodes a 100-entry retention margin. Extract this value into a named constant alongside the existing raftCompactThreshold configuration, then use that shared constant when calculating compactUpTo so startup and runtime compaction margins remain documented and tunable together.drmq-protocol/src/main/proto/messages.proto (1)
42-57: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueOptional:
AtomicBatchRequestandAtomicProduceRequestare structurally identical.Both messages contain only
repeated AtomicBatchTopicSlice slices = 1. This is fine given their distinct roles (Raft entry payload vs. client wire request), but you could collapse them into one type to reduce maintenance surface if the two contracts are expected to stay in lock-step. Leaving them separate is also defensible for future divergence.🤖 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-protocol/src/main/proto/messages.proto` around lines 42 - 57, Optionally consolidate AtomicBatchRequest and AtomicProduceRequest into a single shared message type, updating all references while preserving their distinct protocol roles. If future divergence is expected, retain both definitions and make no change.drmq-broker/src/test/java/com/drmq/broker/MessageStoreTest.java (1)
260-304: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winOptional: add a duplicate-topic test to lock down the contract.
These tests cover the distinct-topic and empty cases well. Given the duplicate-topic data-loss concern flagged in
MessageStore.appendAtomicBatch, a test that passes two slices with the same topic would pin down the intended behavior (reject vs. merge) once that issue is resolved.🤖 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/test/java/com/drmq/broker/MessageStoreTest.java` around lines 260 - 304, Add a test near appendAtomicBatchStoresMessagesAcrossTopics using two AtomicBatchTopicSlice instances with the same topic, and assert the intended appendAtomicBatch behavior once duplicate-topic handling is resolved—either rejecting the input or merging both slices without data loss. Ensure the test verifies the resulting offsets and stored messages according to that contract.drmq-integration-tests/src/test/java/com/drmq/integration/AtomicLatencyBenchmarkTest.java (1)
85-90: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winGate this benchmark out of the default test run.
With 200 warmup + 1000 measured iterations across
{2,4,8}topics — each a real blocking Raft round-trip against a 3-node cluster — this executes as a normal@Testand will substantially slow every CI run. Consider@Disabledwith a comment, or a@Tag("benchmark")/JUnit assumption so it runs only on demand.🤖 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-integration-tests/src/test/java/com/drmq/integration/AtomicLatencyBenchmarkTest.java` around lines 85 - 90, Gate benchmarkAtomicVsSequentialWrites out of the default JUnit test run by applying the project’s supported benchmark mechanism, such as `@Disabled` with an explanatory comment or the established benchmark tag/assumption. Preserve the benchmark’s behavior when explicitly enabled.
🤖 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 192-196: Update addCorsHeaders in AdminHttpServer to remove the
wildcard Access-Control-Allow-Origin policy and enforce authentication/token
validation or allow only the configured dashboard origin. Ensure the restriction
applies to sensitive endpoints including /api/messages and /api/consumers, while
preserving CORS support only for authorized dashboard requests.
- Around line 186-189: Update the exception response handling in
AdminHttpServer’s message-request catch block to construct the error payload
with the existing Gson serializer instead of concatenating e.getMessage() into
JSON. Serialize a structured error value so quotes, newlines, backslashes, and
null messages remain valid JSON, while preserving the 500 response through
sendJsonResponse and avoiding raw internal exception details in the response.
In `@drmq-broker/src/main/java/com/drmq/broker/ClientHandler.java`:
- Around line 232-249: Update the atomic produce validation in ClientHandler to
reject requests with fewer than two slices before selecting the Raft or
message-store path, returning a clear client error consistently in both modes.
Also enforce MAX_BATCH_MESSAGES for the atomic request’s total message count if
that limit is intended to apply, preserving the existing payload-size and
empty-batch checks.
In `@drmq-broker/src/main/java/com/drmq/broker/MessageStore.java`:
- Around line 356-380: Prevent duplicate topics from overwriting data in the
slice-processing flow that builds topicBaseOffsets and topicMessages. Validate
the slices for duplicate topic names before reserving or advancing globalOffset,
and reject the batch atomically with the existing error-handling convention;
otherwise merge same-topic slices consistently before offset assignment so every
message is retained and offsets remain contiguous.
In `@drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java`:
- Around line 1108-1116: The post-commit offset calculation in the proposal
completion path is racy because it reconstructs bases from live head offsets.
Update ProposalState and the appendAtomicBatch/applyCommitted completion flow to
retain and propagate appendAtomicBatch’s authoritative per-topic base-offset
Map, then return that stored map after future.get() instead of iterating slices
and calling messageStore.getHeadOffset.
In `@drmq-broker/src/main/java/com/drmq/broker/raft/SnapshotManager.java`:
- Around line 51-54: Update SnapshotManager’s existing-snapshot handling so it
does not blindly reuse a potentially partial ZIP. Generate the snapshot at a
temporary path, then atomically move it to zipFile only after successful ZIP
creation; alternatively, validate an existing ZIP before returning it and
regenerate when invalid. Preserve the current reuse behavior only for verified
snapshots.
In `@drmq-broker/src/main/java/com/drmq/broker/TelemetryWebSocketServer.java`:
- Around line 159-171: Update updateRates() to include the atomic_produce
counter in both produce-byte/rate aggregations, including the current production
rate and the corresponding produce history values, so atomic-only workloads
contribute to throughput and dashboard records.
In `@drmq-client/src/main/java/com/drmq/client/DRMQProducer.java`:
- Around line 330-352: Update sendAtomic and sendAtomicWithRetry to use a
dedicated atomic executor or thread instead of the common ForkJoinPool, keeping
it separate from the senderThread path used by sendBatchWithRetry. Ensure every
atomic IOException and NOT_LEADER teardown or rotation, including
closeConnection() and rotateToNextServer(), executes while holding sendLock so
shared host, port, socket, in, and out state cannot be changed concurrently.
In `@drmq-dashboard/src/pages/Topics.tsx`:
- Around line 63-70: Update the inspector topic effect and fetchMessages flow so
reopening a topic fetches messages with offset 0 rather than the stale
inspectOffset captured before setInspectOffset completes. Prefer passing the
reset offset explicitly to fetchMessages, while preserving the existing
message-clearing behavior when inspectorTopic is absent.
In `@drmq-dashboard/src/services/telemetry/WebSocketTelemetryProvider.ts`:
- Around line 46-68: The checkConnection retry logic must honor the full
15-second window even when sockets are closed or null while scheduleReconnect()
is pending. Update checkConnection so it reschedules checks until Date.now() -
startTime reaches 15000, regardless of anyConnecting; only invoke
onErrorCallback after that deadline, while preserving the stopped and anyOpen
early returns.
In `@drmq-ts-client/src/client.ts`:
- Around line 346-385: Prevent the non-retryable error raised in the atomic
produce retry loop from being swallowed by its surrounding catch. Update the
atomic produce method’s error path to preserve and surface the fatal broker
error after the retry handling, while retaining connection rotation and retries
only for retryable failures; ensure the original “Failed to send atomic batch”
message is returned immediately instead of becoming a timeout.
---
Nitpick comments:
In `@drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.java`:
- Around line 144-152: Update the query parsing in AdminHttpServer to split each
parameter only at its first “=” and URL-decode both parameter names and values
before matching topic, offset, limit, and timestamp. Catch malformed numeric
values from Long.parseLong and Integer.parseInt, and return an HTTP 400 response
instead of allowing NumberFormatException to surface as a 500.
In `@drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java`:
- Around line 1551-1557: The startup compaction logic in RaftNode currently
hardcodes a 100-entry retention margin. Extract this value into a named constant
alongside the existing raftCompactThreshold configuration, then use that shared
constant when calculating compactUpTo so startup and runtime compaction margins
remain documented and tunable together.
In `@drmq-broker/src/test/java/com/drmq/broker/MessageStoreTest.java`:
- Around line 260-304: Add a test near
appendAtomicBatchStoresMessagesAcrossTopics using two AtomicBatchTopicSlice
instances with the same topic, and assert the intended appendAtomicBatch
behavior once duplicate-topic handling is resolved—either rejecting the input or
merging both slices without data loss. Ensure the test verifies the resulting
offsets and stored messages according to that contract.
In `@drmq-dashboard/src/pages/Topics.tsx`:
- Line 77: Replace the hardcoded admin origin in Topics.tsx, including both
fetchTopics and the /api/messages request, and Consumers.tsx in fetchConsumers
with one shared, environment- or telemetry-driven admin base URL. Define or
reuse the shared configuration so all three requests resolve against the
intended broker rather than localhost:9392.
In
`@drmq-integration-tests/src/test/java/com/drmq/integration/AtomicLatencyBenchmarkTest.java`:
- Around line 85-90: Gate benchmarkAtomicVsSequentialWrites out of the default
JUnit test run by applying the project’s supported benchmark mechanism, such as
`@Disabled` with an explanatory comment or the established benchmark
tag/assumption. Preserve the benchmark’s behavior when explicitly enabled.
In `@drmq-protocol/src/main/proto/messages.proto`:
- Around line 42-57: Optionally consolidate AtomicBatchRequest and
AtomicProduceRequest into a single shared message type, updating all references
while preserving their distinct protocol roles. If future divergence is
expected, retain both definitions and make no 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 Plus
Run ID: d60250e7-6b9f-41a9-95ee-3985805aaf16
📒 Files selected for processing (27)
drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.javadrmq-broker/src/main/java/com/drmq/broker/BrokerConfig.javadrmq-broker/src/main/java/com/drmq/broker/BrokerServer.javadrmq-broker/src/main/java/com/drmq/broker/ClientHandler.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/RaftLog.javadrmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.javadrmq-broker/src/main/java/com/drmq/broker/raft/SnapshotManager.javadrmq-broker/src/test/java/com/drmq/broker/MessageStoreTest.javadrmq-client/src/main/java/com/drmq/client/DRMQProducer.javadrmq-client/src/main/java/com/drmq/client/commandLineExample/AtomicStressTestApp.javadrmq-client/src/main/java/com/drmq/client/commandLineExample/ProducerApp.javadrmq-dashboard/src/App.tsxdrmq-dashboard/src/pages/Consumers.tsxdrmq-dashboard/src/pages/Dashboard.tsxdrmq-dashboard/src/pages/Topics.tsxdrmq-dashboard/src/services/telemetry/WebSocketTelemetryProvider.tsdrmq-integration-tests/src/test/java/com/drmq/integration/AtomicLatencyBenchmarkTest.javadrmq-protocol/src/main/proto/messages.protodrmq-python-client/drmq_client.pydrmq-python-client/messages_pb2.pydrmq-python-client/test_time_seek.pydrmq-ts-client/src/client.tsdrmq-ts-client/src/messages.ts
💤 Files with no reviewable changes (1)
- drmq-broker/src/main/java/com/drmq/broker/ConsumerGroupCoordinator.java
| } 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 | 🟠 Major | ⚡ Quick win
Build the error JSON with Gson instead of string concatenation.
e.getMessage() is interpolated raw into the JSON body. If the message contains a ", newline, or backslash the response becomes malformed JSON and the dashboard's res.json() fails; if it is null the body reads {"error":"null"}. It also leaks internal exception detail verbatim.
🐛 Proposed fix
} 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() != null ? e.getMessage() : "Internal error");
+ 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() != null ? e.getMessage() : "Internal error"); | |
| 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 handling in AdminHttpServer’s
message-request catch block to construct the error payload with the existing
Gson serializer instead of concatenating e.getMessage() into JSON. Serialize a
structured error value so quotes, newlines, backslashes, and null messages
remain valid JSON, while preserving the 500 response through sendJsonResponse
and avoiding raw internal exception details in the response.
| 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"); | ||
| } |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm whether any auth/token check gates these admin endpoints elsewhere.
rg -nP 'Authorization|Bearer|token|apiKey|AdminHttpServer' drmq-broker/src/main/java --type=java -C2Repository: samuel025/DRMQ
Length of output: 3093
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect AdminHttpServer handlers and related message endpoints/cors behavior.
wc -l drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.java
cat -n drmq-broker/src/main/java/com/drmq/broker/AdminHttpServer.java | sed -n '1,280p'
# Search for /api/messages/path, getPayload, or raw message serialization in Java sources.
rg -n '/api/messages|messages|getPayload|payload|Access-Control-Allow-Origin|addCorsHeaders|sendResponse' drmq-broker/src/main/java drmq-broker/src/main -g '*.java' -C2Repository: samuel025/DRMQ
Length of output: 50371
Restrict admin CORS or require authentication.
/api/messages returns raw message payloads, and the admin server sets Access-Control-Allow-Origin: "*". Any page an operator visits can treat localhost:<adminPort> as same-origin for data reads; also /api/consumers exposes consumer/lag state. Require auth/token or restrict CORS to the dashboard origin instead of allowing cross-origin reads from every origin.
🤖 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
192 - 196, Update addCorsHeaders in AdminHttpServer to remove the wildcard
Access-Control-Allow-Origin policy and enforce authentication/token validation
or allow only the configured dashboard origin. Ensure the restriction applies to
sensitive endpoints including /api/messages and /api/consumers, while preserving
CORS support only for authorized dashboard requests.
| if (batchCount == 0) { | ||
| return createAtomicProduceErrorResponse("Atomic batch must contain at least one message", ErrorCode.UNKNOWN_ERROR); | ||
| } | ||
| if (totalPayloadBytes > MAX_PAYLOAD_BYTES) { | ||
| return createAtomicProduceErrorResponse("Batch payload exceeds maximum size of " + MAX_PAYLOAD_BYTES + " bytes", ErrorCode.UNKNOWN_ERROR); | ||
| } | ||
|
|
||
| java.util.Map<String, Long> offsets; | ||
| if (raftNode != null) { | ||
| if (!raftNode.isLeader()) { | ||
| String leaderAddr = raftNode.getLeaderAddress(); | ||
| return createAtomicProduceErrorResponse("NOT_LEADER:" + | ||
| (leaderAddr != null ? leaderAddr : "UNKNOWN"), ErrorCode.NOT_LEADER); | ||
| } | ||
| offsets = raftNode.proposeAtomicBatch(request.getSlicesList()); | ||
| } else { | ||
| offsets = messageStore.appendAtomicBatch(request.getSlicesList()); | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Inconsistent handling of single-slice atomic requests between cluster and single-node paths.
There is no explicit "at least 2 topics" check here. In cluster mode raftNode.proposeAtomicBatch throws IllegalArgumentException for <2 slices, which surfaces as a generic UNKNOWN_ERROR; in single-node mode messageStore.appendAtomicBatch accepts a single slice. Add an up-front validation returning a clear error so both paths behave the same regardless of Raft mode. Also note MAX_BATCH_MESSAGES is enforced for PRODUCE_BATCH_REQUEST but not here — confirm whether the atomic path should bound message count too.
🤖 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/ClientHandler.java` around lines
232 - 249, Update the atomic produce validation in ClientHandler to reject
requests with fewer than two slices before selecting the Raft or message-store
path, returning a clear client error consistently in both modes. Also enforce
MAX_BATCH_MESSAGES for the atomic request’s total message count if that limit is
intended to apply, preserving the existing payload-size and empty-batch checks.
| Map<String, Long> topicBaseOffsets = new LinkedHashMap<>(); | ||
| Map<String, List<StoredMessage>> topicMessages = new LinkedHashMap<>(); | ||
|
|
||
| long currentOffset = baseOffset; | ||
| for (var slice : slices) { | ||
| String topic = slice.getTopic(); | ||
| long topicBase = currentOffset; | ||
| topicBaseOffsets.put(topic, topicBase); | ||
|
|
||
| List<StoredMessage> messages = new ArrayList<>(slice.getEntriesCount()); | ||
| for (var entry : slice.getEntriesList()) { | ||
| StoredMessage.Builder builder = StoredMessage.newBuilder() | ||
| .setOffset(currentOffset++) | ||
| .setTopic(topic) | ||
| .setPayload(entry.getPayload()) | ||
| .setTimestamp(entry.getClientTimestamp()) | ||
| .setStoredAt(storedAt); | ||
|
|
||
| if (entry.hasKey()) { | ||
| builder.setKey(entry.getKey()); | ||
| } | ||
| messages.add(builder.build()); | ||
| } | ||
| topicMessages.put(topic, messages); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Duplicate topics across slices silently drop messages and create offset gaps.
topicBaseOffsets and topicMessages are keyed by topic in a LinkedHashMap. If two slices share the same topic, the second put overwrites the first, so the first slice's StoredMessage list is discarded — but globalOffset was already advanced by totalMessages (line 353). The result is permanently lost messages and a gap in the global offset space. The downstream offset derivation in RaftNode.proposeAtomicBatch (getHeadOffset - entriesCount + 1) is also wrong under duplicate topics.
The provided clients pass a map so duplicates can't originate there, but the broker API accepts a List and should defend the atomicity contract itself. Consider rejecting duplicate topics up front, or merging same-topic slices before reserving offsets.
🤖 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 356
- 380, Prevent duplicate topics from overwriting data in the slice-processing
flow that builds topicBaseOffsets and topicMessages. Validate the slices for
duplicate topic names before reserving or advancing globalOffset, and reject the
batch atomically with the existing error-handling convention; otherwise merge
same-topic slices consistently before offset assignment so every message is
retained and offsets remain contiguous.
| try { | ||
| future.get(PROPOSAL_TIMEOUT_SECONDS, TimeUnit.SECONDS); | ||
| // After commit, fetch offsets from the message store | ||
| Map<String, Long> offsets = new java.util.LinkedHashMap<>(); | ||
| for (com.drmq.protocol.DRMQProtocol.AtomicBatchTopicSlice s : slices) { | ||
| offsets.put(s.getTopic(), | ||
| messageStore.getHeadOffset(s.getTopic()) - s.getEntriesCount() + 1); | ||
| } | ||
| return offsets; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Post-commit base-offset derivation is racy and can return wrong offsets.
After future.get(), offsets are recomputed as getHeadOffset(topic) - entriesCount + 1. applyCommitted runs on other threads, so between this future completing and the loop below, another committed entry that writes the same topic can advance the head offset — yielding a base offset that is too high for the client. MessageStore.appendAtomicBatch already computes and returns the exact per-topic base offsets; that authoritative value is discarded here and re-derived unsafely.
Prefer plumbing the actual Map<String,Long> base offsets from appendAtomicBatch through the proposal completion (e.g. store it on the ProposalState) rather than reconstructing from live head offsets.
🤖 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
1108 - 1116, The post-commit offset calculation in the proposal completion path
is racy because it reconstructs bases from live head offsets. Update
ProposalState and the appendAtomicBatch/applyCommitted completion flow to retain
and propagate appendAtomicBatch’s authoritative per-topic base-offset Map, then
return that stored map after future.get() instead of iterating slices and
calling messageStore.getHeadOffset.
| double totalMBps = currentProduceMBps + currentConsumeMBps; | ||
| // Scale: 50 MB/s = 100 on chart | ||
| double chartVal = Math.min(100, (totalMBps / 50.0) * 100); | ||
| if (chartVal < 3 && totalMBps > 0) chartVal = 3; // minimum visibility | ||
| synchronized (throughputHistory) { | ||
| throughputHistory.remove(0); | ||
| throughputHistory.add(chartVal); | ||
| throughputHistory.add(totalMBps); | ||
|
|
||
| produceHistory.remove(0); | ||
| produceHistory.add(currentProduceMBps); | ||
|
|
||
| consumeHistory.remove(0); | ||
| consumeHistory.add(currentConsumeMBps); | ||
|
|
||
| errorHistory.remove(0); | ||
| errorHistory.add(currentErrorRate); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Include atomic production in rate aggregation.
ClientHandler.handleAtomicProduceRequest() records atomic_produce, but updateRates() excludes it from produce bytes and records. Atomic-only workloads therefore display zero throughput and rate across these new histories and the dashboard.
Add atomic_produce to both produce counter sums.
🤖 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/TelemetryWebSocketServer.java`
around lines 159 - 171, Update updateRates() to include the atomic_produce
counter in both produce-byte/rate aggregations, including the current production
rate and the corresponding produce history values, so atomic-only workloads
contribute to throughput and dashboard records.
| public CompletableFuture<java.util.Map<String, Long>> sendAtomic(java.util.Map<String, byte[]> topicMessages) { | ||
| CompletableFuture<java.util.Map<String, Long>> future = new CompletableFuture<>(); | ||
|
|
||
| if (topicMessages.size() < 2) { | ||
| future.completeExceptionally(new IllegalArgumentException("sendAtomic() requires at least 2 topics")); | ||
| return future; | ||
| } | ||
|
|
||
| com.drmq.protocol.DRMQProtocol.AtomicProduceRequest.Builder reqBuilder = com.drmq.protocol.DRMQProtocol.AtomicProduceRequest.newBuilder(); | ||
| for (java.util.Map.Entry<String, byte[]> e : topicMessages.entrySet()) { | ||
| reqBuilder.addSlices(com.drmq.protocol.DRMQProtocol.AtomicBatchTopicSlice.newBuilder() | ||
| .setTopic(e.getKey()) | ||
| .addEntries(com.drmq.protocol.DRMQProtocol.ProduceBatchRequest.BatchEntry.newBuilder() | ||
| .setPayload(com.google.protobuf.ByteString.copyFrom(e.getValue())) | ||
| .setClientTimestamp(System.currentTimeMillis()) | ||
| .build()) | ||
| .build()); | ||
| } | ||
|
|
||
| AtomicProduceRequest request = reqBuilder.build(); | ||
| CompletableFuture.runAsync(() -> sendAtomicWithRetry(request, future)); | ||
| return future; | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm host/port/socket field declarations (volatility) and every mutation site.
rg -nP 'private\s+(volatile\s+)?(String\s+host|int\s+port|Socket\s+socket)' drmq-client/src/main/java/com/drmq/client/DRMQProducer.java
rg -nP '\b(closeConnection|rotateToNextServer|redirectToLeader)\s*\(' drmq-client/src/main/java/com/drmq/client/DRMQProducer.javaRepository: samuel025/DRMQ
Length of output: 1595
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== file size and relevant references =="
wc -l drmq-client/src/main/java/com/drmq/client/DRMQProducer.java
rg -n -C 3 'sendBatchWithRetry|sendAtomic|sendAtomicWithRetry|synchronized \(sendLock\)|senderThread|Thread\.sleep|sendLock|closeConnection\(|rotateToNextServer\(|redirectToLeader\(' drmq-client/src/main/java/com/drmq/client/DRMQProducer.java
echo "== outline =="
ast-grep outline drmq-client/src/main/java/com/drmq/client/DRMQProducer.java --view compact || true
echo "== targeted ranges =="
sed -n '1,150p' drmq-client/src/main/java/com/drmq/client/DRMQProducer.java | cat -n
sed -n '230,455p' drmq-client/src/main/java/com/drmq/client/DRMQProducer.java | cat -n
sed -n '455,520p' drmq-client/src/main/java/com/drmq/client/DRMQProducer.java | cat -nRepository: samuel025/DRMQ
Length of output: 33400
Protect the atomic send from competing with the dedicated sender path.
sendAtomic runs sendAtomicWithRetry on the common ForkJoinPool, while sendBatchWithRetry continues to drain on senderThread using the same host/port/socket/in/out fields. The IOException and NOT_LEADER branches call closeConnection()/rotateToNextServer() outside sendLock, and those fields are non-volatile, so an atomic retry can rotate the connection target while the sender thread is already reading or writing, breaking normal sends. Use a dedicated atomic executor/thread path and perform all connection teardown/rotation under sendLock.
🤖 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-client/src/main/java/com/drmq/client/DRMQProducer.java` around lines 330
- 352, Update sendAtomic and sendAtomicWithRetry to use a dedicated atomic
executor or thread instead of the common ForkJoinPool, keeping it separate from
the senderThread path used by sendBatchWithRetry. Ensure every atomic
IOException and NOT_LEADER teardown or rotation, including closeConnection() and
rotateToNextServer(), executes while holding sendLock so shared host, port,
socket, in, and out state cannot be changed concurrently.
| 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 | 🟡 Minor | ⚡ Quick win
Stale closure: fetchMessages() runs with the previous inspectOffset.
setInspectOffset(0) is asynchronous, so the immediately-following fetchMessages() still closes over the prior inspectOffset value. When reopening a topic whose offset was previously snapped to a non-zero value, the first fetch starts from that stale offset instead of 0. Consider passing an explicit offset argument to fetchMessages or fetching from a follow-up effect keyed on inspectOffset.
🤖 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 63 - 70, Update the
inspector topic effect and fetchMessages flow so reopening a topic fetches
messages with offset 0 rather than the stale inspectOffset captured before
setInspectOffset completes. Prefer passing the reset offset explicitly to
fetchMessages, while preserving the existing message-clearing behavior when
inspectorTopic is absent.
|
|
||
| // Retry-aware connection check: don't show timeout error if sockets are | ||
| // still actively handshaking (CONNECTING). Only give up after 15s total. | ||
| const startTime = Date.now(); | ||
| const checkConnection = () => { | ||
| if (this.stopped) return; | ||
| const sockets = Array.from(this.sockets.values()); | ||
| const anyOpen = sockets.some(s => s && s.readyState === WebSocket.OPEN); | ||
| if (anyOpen) return; // At least one connected — all good | ||
|
|
||
| const anyConnecting = sockets.some(s => s && s.readyState === WebSocket.CONNECTING); | ||
| if (anyConnecting && Date.now() - startTime < 15000) { | ||
| // Still handshaking — check again in 2s instead of giving up | ||
| setTimeout(checkConnection, 2000); | ||
| return; | ||
| } | ||
|
|
||
| // All sockets are closed/null or we've waited 15s — show error | ||
| if (this.onErrorCallback) { | ||
| this.onErrorCallback('Connection timeout. Check if brokers are running.'); | ||
| } | ||
| }, 5000); | ||
| }; | ||
| setTimeout(checkConnection, 5000); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Honor the full 15-second retry window.
A refused connection transitions to closed/null while scheduleReconnect() is pending, so anyConnecting is false and this emits a timeout after 5 seconds. Retry until Date.now() - startTime >= 15000 regardless of the current socket state; only then report the timeout.
🤖 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/services/telemetry/WebSocketTelemetryProvider.ts` around
lines 46 - 68, The checkConnection retry logic must honor the full 15-second
window even when sockets are closed or null while scheduleReconnect() is
pending. Update checkConnection so it reschedules checks until Date.now() -
startTime reaches 15000, regardless of anyConnecting; only invoke
onErrorCallback after that deadline, while preserving the stopped and anyOpen
early returns.
| try { | ||
| await this.ensureConnected(); | ||
| const respBytes = await this.sendEnvelope(MessageType.ATOMIC_PRODUCE_REQUEST, envelopeBytes); | ||
| const resp = AtomicProduceResponse.decode(respBytes); | ||
|
|
||
| if (resp.success) { | ||
| const result: Record<string, number> = {}; | ||
| for (const [k, v] of Object.entries(resp.baseOffsets)) { | ||
| result[k] = Number(v); | ||
| } | ||
| return result; | ||
| } else { | ||
| const errorMsg = resp.errorMessage; | ||
| if (resp.errorCode === ErrorCode.NOT_LEADER || (errorMsg && errorMsg.startsWith("NOT_LEADER:"))) { | ||
| const leaderAddr = errorMsg && errorMsg.startsWith("NOT_LEADER:") | ||
| ? errorMsg.substring("NOT_LEADER:".length) | ||
| : "UNKNOWN"; | ||
| if (leaderAddr !== "UNKNOWN") { | ||
| const parts = leaderAddr.split(":"); | ||
| if (parts.length === 2) { | ||
| this.host = parts[0]; | ||
| this.port = parseInt(parts[1], 10); | ||
| super.closeConnection(); | ||
| await this.ensureConnected(); | ||
| continue; | ||
| } | ||
| } | ||
| super.closeConnection(); | ||
| this.rotateServer(); | ||
| } else if (errorMsg && (errorMsg.includes("timed out") || errorMsg.includes("Lost leadership") || errorMsg.includes("Raft batch proposal"))) { | ||
| super.closeConnection(); | ||
| this.rotateServer(); | ||
| } else { | ||
| throw new Error(`Failed to send atomic batch: ${errorMsg}`); | ||
| } | ||
| } | ||
| } catch (e) { | ||
| super.closeConnection(); | ||
| this.rotateServer(); | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Non-retryable error is swallowed by the surrounding catch, turning fail-fast into a 120s timeout.
The throw new Error(\Failed to send atomic batch: ${errorMsg}`)on Line 379 is inside the sametrywhosecatch (e)(Lines 382-385) closes the connection and rotates the server. So a genuinely non-retryable broker error is caught, retried untildeliveryTimeoutMs, and finally reported as "Failed to send atomic batch: timeout exhausted"— the real error message is lost. This diverges fromsendBatchWithRetry(Lines 301-304, whichresolve(...)+return`) and from the Java/Python clients (which complete/set the future and return). Fail fast on non-retryable errors instead.
🐛 Proposed fix: surface the fatal error after the loop instead of throwing inside try
const startMs = Date.now();
let currentBackoffMs = 100;
+ let fatalError: Error | null = null;
while (true) {
if (Date.now() - startMs > deliveryTimeoutMs) {
break;
}
@@
} else if (errorMsg && (errorMsg.includes("timed out") || errorMsg.includes("Lost leadership") || errorMsg.includes("Raft batch proposal"))) {
super.closeConnection();
this.rotateServer();
} else {
- throw new Error(`Failed to send atomic batch: ${errorMsg}`);
+ fatalError = new Error(`Failed to send atomic batch: ${errorMsg}`);
+ break;
}
}
} catch (e) {
+ if (fatalError) break;
super.closeConnection();
this.rotateServer();
}
+ if (fatalError) break;
await new Promise(res => setTimeout(res, currentBackoffMs));
currentBackoffMs = Math.min(2000, currentBackoffMs * 2);
}
+ if (fatalError) throw fatalError;
throw new Error("Failed to send atomic batch: timeout exhausted");📝 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.
| try { | |
| await this.ensureConnected(); | |
| const respBytes = await this.sendEnvelope(MessageType.ATOMIC_PRODUCE_REQUEST, envelopeBytes); | |
| const resp = AtomicProduceResponse.decode(respBytes); | |
| if (resp.success) { | |
| const result: Record<string, number> = {}; | |
| for (const [k, v] of Object.entries(resp.baseOffsets)) { | |
| result[k] = Number(v); | |
| } | |
| return result; | |
| } else { | |
| const errorMsg = resp.errorMessage; | |
| if (resp.errorCode === ErrorCode.NOT_LEADER || (errorMsg && errorMsg.startsWith("NOT_LEADER:"))) { | |
| const leaderAddr = errorMsg && errorMsg.startsWith("NOT_LEADER:") | |
| ? errorMsg.substring("NOT_LEADER:".length) | |
| : "UNKNOWN"; | |
| if (leaderAddr !== "UNKNOWN") { | |
| const parts = leaderAddr.split(":"); | |
| if (parts.length === 2) { | |
| this.host = parts[0]; | |
| this.port = parseInt(parts[1], 10); | |
| super.closeConnection(); | |
| await this.ensureConnected(); | |
| continue; | |
| } | |
| } | |
| super.closeConnection(); | |
| this.rotateServer(); | |
| } else if (errorMsg && (errorMsg.includes("timed out") || errorMsg.includes("Lost leadership") || errorMsg.includes("Raft batch proposal"))) { | |
| super.closeConnection(); | |
| this.rotateServer(); | |
| } else { | |
| throw new Error(`Failed to send atomic batch: ${errorMsg}`); | |
| } | |
| } | |
| } catch (e) { | |
| super.closeConnection(); | |
| this.rotateServer(); | |
| } | |
| const startMs = Date.now(); | |
| let currentBackoffMs = 100; | |
| let fatalError: Error | null = null; | |
| while (true) { | |
| if (Date.now() - startMs > deliveryTimeoutMs) { | |
| break; | |
| } | |
| try { | |
| await this.ensureConnected(); | |
| const respBytes = await this.sendEnvelope(MessageType.ATOMIC_PRODUCE_REQUEST, envelopeBytes); | |
| const resp = AtomicProduceResponse.decode(respBytes); | |
| if (resp.success) { | |
| const result: Record<string, number> = {}; | |
| for (const [k, v] of Object.entries(resp.baseOffsets)) { | |
| result[k] = Number(v); | |
| } | |
| return result; | |
| } else { | |
| const errorMsg = resp.errorMessage; | |
| if (resp.errorCode === ErrorCode.NOT_LEADER || (errorMsg && errorMsg.startsWith("NOT_LEADER:"))) { | |
| const leaderAddr = errorMsg && errorMsg.startsWith("NOT_LEADER:") | |
| ? errorMsg.substring("NOT_LEADER:".length) | |
| : "UNKNOWN"; | |
| if (leaderAddr !== "UNKNOWN") { | |
| const parts = leaderAddr.split(":"); | |
| if (parts.length === 2) { | |
| this.host = parts[0]; | |
| this.port = parseInt(parts[1], 10); | |
| super.closeConnection(); | |
| await this.ensureConnected(); | |
| continue; | |
| } | |
| } | |
| super.closeConnection(); | |
| this.rotateServer(); | |
| } else if (errorMsg && (errorMsg.includes("timed out") || errorMsg.includes("Lost leadership") || errorMsg.includes("Raft batch proposal"))) { | |
| super.closeConnection(); | |
| this.rotateServer(); | |
| } else { | |
| fatalError = new Error(`Failed to send atomic batch: ${errorMsg}`); | |
| break; | |
| } | |
| } | |
| } catch (e) { | |
| if (fatalError) break; | |
| super.closeConnection(); | |
| this.rotateServer(); | |
| } | |
| if (fatalError) break; | |
| await new Promise(res => setTimeout(res, currentBackoffMs)); | |
| currentBackoffMs = Math.min(2000, currentBackoffMs * 2); | |
| } | |
| if (fatalError) throw fatalError; | |
| throw new Error("Failed to send atomic batch: timeout exhausted"); |
🤖 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-ts-client/src/client.ts` around lines 346 - 385, Prevent the
non-retryable error raised in the atomic produce retry loop from being swallowed
by its surrounding catch. Update the atomic produce method’s error path to
preserve and surface the fatal broker error after the retry handling, while
retaining connection rotation and retries only for retryable failures; ensure
the original “Failed to send atomic batch” message is returned immediately
instead of becoming a timeout.
1 participant
Added cross-topic atomicity for production
Summary by CodeRabbit