Feat/log compaction

.mvn/wrapper/maven-wrapper.properties

@@ -0,0 +1,4 @@
+wrapperVersion=3.3.4
+distributionType=only-script
+distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.8.7/apache-maven-3.8.7-bin.zip
+distributionSha512Sum=1109a1a89c93dd4195a6fcc8cba279fb08e4d3a017088b90c1f54b6fb7d3ceccf8539eecb08fccba6462bb8bba99af2f073bb8956bb201887eabcc843a60a7e6

README.md

@@ -49,7 +49,7 @@ mvn clean install
 To run a standalone broker (useful for testing and development):
 
 ```bash
-java -jar drmq-broker/target/drmq-broker-1.0.0-SNAPSHOT.jar --port 9092 --data-dir ./data-1
+./mvnw -pl drmq-broker exec:java -Dexec.args="--port 9092 --data-dir ./data-1"
 ```
 
 ### Cluster Mode
@@ -59,19 +59,19 @@ To run a fault-tolerant cluster, you must start multiple broker instances and pr
 **Node 1:**
 
 ```bash
-java -jar drmq-broker/target/drmq-broker-1.0.0-SNAPSHOT.jar --node-id 1 --port 9092 --data-dir ./data-1 --peers 2:localhost:9093,3:localhost:9094
+./mvnw -pl drmq-broker exec:java -Dexec.args="--node-id 1 --port 9092 --data-dir ./data-1 --peers 2:localhost:9093,3:localhost:9094"
 ```
 
 **Node 2:**
 
 ```bash
-java -jar drmq-broker/target/drmq-broker-1.0.0-SNAPSHOT.jar --node-id 2 --port 9093 --data-dir ./data-2 --peers 1:localhost:9092,3:localhost:9094
+./mvnw -pl drmq-broker exec:java -Dexec.args="--node-id 2 --port 9093 --data-dir ./data-2 --peers 1:localhost:9092,3:localhost:9094"
 ```
 
 **Node 3:**
 
 ```bash
-java -jar drmq-broker/target/drmq-broker-1.0.0-SNAPSHOT.jar --node-id 3 --port 9094 --data-dir ./data-3 --peers 1:localhost:9092,2:localhost:9093
+./mvnw -pl drmq-broker exec:java -Dexec.args="--node-id 3 --port 9094 --data-dir ./data-3 --peers 1:localhost:9092,2:localhost:9093"
 ```
 
 ## Usage Example

drmq-broker/pom.xml

@@ -45,6 +45,16 @@
             <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.java-websocket</groupId>
+            <artifactId>Java-WebSocket</artifactId>
+            <version>1.5.3</version>
+        </dependency>
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+            <version>2.10.1</version>
+        </dependency>
     </dependencies>
 
     <build>

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

@@ -19,25 +19,34 @@ public class BrokerConfig {
     private final boolean metricsEnabled;
     private final int metricsPort;
     private final String metricsPath;
+    private long logSegmentBytes;
+    private long logRetentionMs;
+    private final long raftCompactThreshold;
 
     public BrokerConfig(String nodeId, int port, String dataDir, List<PeerAddress> peers,
-                        boolean metricsEnabled, int metricsPort, String metricsPath) {
+                        boolean metricsEnabled, int metricsPort, String metricsPath,
+                        long logSegmentBytes, long logRetentionMs, long raftCompactThreshold) {
         this.nodeId = nodeId;
         this.port = port;
         this.dataDir = dataDir;
         this.peers = peers != null ? peers : List.of();
         this.metricsEnabled = metricsEnabled;
         this.metricsPort = metricsPort;
         this.metricsPath = metricsPath != null ? metricsPath : "/metrics";
+        this.logSegmentBytes = logSegmentBytes;
+        this.logRetentionMs = logRetentionMs;
+        this.raftCompactThreshold = raftCompactThreshold;
     }
 
     public BrokerConfig(String nodeId, int port, String dataDir, List<PeerAddress> peers) {
-        this(nodeId, port, dataDir, peers, true, 9096, "/metrics");
+        this(nodeId, port, dataDir, peers, true, 9096, "/metrics", 
+             100 * 1024 * 1024L, 7L * 24 * 60 * 60 * 1000, 1000L);
     }
 
     /** Single-node config (backward compatible) */
     public BrokerConfig(int port, String dataDir) {
-        this("standalone", port, dataDir, List.of(), true, 9096, "/metrics");
+        this("standalone", port, dataDir, List.of(), true, 9096, "/metrics",
+             100 * 1024 * 1024L, 7L * 24 * 60 * 60 * 1000, 1000L);
     }
 
     public String getNodeId() { return nodeId; }
@@ -47,6 +56,23 @@ public BrokerConfig(int port, String dataDir) {
     public boolean isMetricsEnabled() { return metricsEnabled; }
     public int getMetricsPort() { return metricsPort; }
     public String getMetricsPath() { return metricsPath; }
+    public long getLogSegmentBytes() { return logSegmentBytes; }
+    public long getLogRetentionMs() { return logRetentionMs; }
+    public long getRaftCompactThreshold() { return raftCompactThreshold; }
+
+    public void setLogSegmentBytes(long logSegmentBytes) {
+        if (logSegmentBytes <= 0) {
+            throw new IllegalArgumentException("logSegmentBytes must be positive");
+        }
+        this.logSegmentBytes = logSegmentBytes;
+    }
+
+    public void setLogRetentionMs(long logRetentionMs) {
+        if (logRetentionMs <= 0) {
+            throw new IllegalArgumentException("logRetentionMs must be positive");
+        }
+        this.logRetentionMs = logRetentionMs;
+    }
 
     /** True if this broker is part of a Raft cluster */
     public boolean isClusterMode() {
@@ -65,6 +91,9 @@ public boolean isClusterMode() {
      *   --metrics-disabled
      *   --metrics-port <port>
      *   --metrics-path </metrics>
+     *   --log-segment-bytes <bytes>
+     *   --log-retention-ms <ms>
+     *   --raft-compact-threshold <count>
      */
     public static BrokerConfig fromArgs(String[] args) {
         String nodeId = "standalone";
@@ -74,10 +103,13 @@ public static BrokerConfig fromArgs(String[] args) {
         boolean metricsEnabled = true;
         int metricsPort = 9096;
         String metricsPath = "/metrics";
+        long logSegmentBytes = 100 * 1024 * 1024L; // 100MB
+        long logRetentionMs = 7L * 24 * 60 * 60 * 1000; // 7 days
+        long raftCompactThreshold = 1000L;
 
         for (int i = 0; i < args.length; i++) {
             switch (args[i]) {
-                case "--id" -> nodeId = args[++i];
+                case "--id", "--node-id" -> nodeId = args[++i];
                 case "--port" -> port = Integer.parseInt(args[++i]);
                 case "--data-dir" -> dataDir = args[++i];
                 case "--peers" -> {
@@ -90,6 +122,9 @@ public static BrokerConfig fromArgs(String[] args) {
                 case "--metrics-disabled" -> metricsEnabled = false;
                 case "--metrics-port" -> metricsPort = parsePortArg(args, ++i, "--metrics-port");
                 case "--metrics-path" -> metricsPath = parsePathArg(args, ++i, "--metrics-path");
+                case "--log-segment-bytes" -> logSegmentBytes = parseLongArg(args, ++i, "--log-segment-bytes");
+                case "--log-retention-ms" -> logRetentionMs = parseLongArg(args, ++i, "--log-retention-ms");
+                case "--raft-compact-threshold" -> raftCompactThreshold = parseLongArg(args, ++i, "--raft-compact-threshold");
                 default -> {
                     if (i == 0) {
                         try {
@@ -104,7 +139,8 @@ public static BrokerConfig fromArgs(String[] args) {
             }
         }
 
-        return new BrokerConfig(nodeId, port, dataDir, peers, metricsEnabled, metricsPort, metricsPath);
+        return new BrokerConfig(nodeId, port, dataDir, peers, metricsEnabled, metricsPort, metricsPath,
+                                logSegmentBytes, logRetentionMs, raftCompactThreshold);
     }
 
     private static boolean parseBooleanArg(String[] args, int index, String flag) {
@@ -131,6 +167,19 @@ private static int parsePortArg(String[] args, int index, String flag) {
         }
     }
 
+    private static long parseLongArg(String[] args, int index, String flag) {
+        String value = requireValue(args, index, flag);
+        try {
+            long parsed = Long.parseLong(value);
+            if (parsed <= 0) {
+                throw new IllegalArgumentException(flag + " must be positive");
+            }
+            return parsed;
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException(flag + " must be a valid long integer, got: " + value, e);
+        }
+    }
+
     private static String parsePathArg(String[] args, int index, String flag) {
         String value = requireValue(args, index, flag);
         String normalized = value.trim();

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

@@ -133,6 +133,14 @@ public void registerBroker(java.util.function.Supplier<Integer> activeConnection
                     .register(registry);
             Gauge.builder("drmq.broker.raft.is_leader", raftNode, node -> node.isLeader() ? 1 : 0)
                     .register(registry);
+            for (String peerId : raftNode.getPeerIds()) {
+                Gauge.builder("drmq.broker.raft.replication_lag", raftNode, node -> {
+                    if (!node.isLeader()) return 0.0;
+                    Long matchIdx = node.getMatchIndexMap().get(peerId);
+                    if (matchIdx == null) return 0.0;
+                    return (double) (node.getLastLogIndex() - matchIdx);
+                }).tags("peer_id", peerId).register(registry);
+            }
         }
     }
 
@@ -354,6 +362,35 @@ private void appendJsonLatency(StringBuilder json, String key, double meanSecond
         json.append('}');
     }
 
+    public double getCounterValueByType(String name, String type) {
+        if (!enabled || registry == null) {
+            return 0.0;
+        }
+        return counterValue(name, Tags.of("type", type));
+    }
+
+    /**
+     * Return the current value of a named gauge, or 0 if metrics are disabled or the gauge doesn't exist.
+     */
+    public double getGaugeValue(String name) {
+        if (!enabled || registry == null) {
+            return 0.0;
+        }
+        var gauge = registry.find(name).gauge();
+        return gauge != null ? gauge.value() : 0.0;
+    }
+
+    /**
+     * Return the mean duration (in milliseconds) of a named timer tagged by type,
+     * or 0 if no samples have been recorded.
+     */
+    public double getTimerMeanMs(String name, String type) {
+        if (!enabled || registry == null) {
+            return 0.0;
+        }
+        return timerMeanSeconds(name, Tags.of("type", type)) * 1000.0;
+    }
+
     private double counterValue(String name, Tags tags) {
         Counter counter = registry.find(name).tags(tags).counter();
         return counter != null ? counter.count() : 0.0;

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

@@ -41,6 +41,7 @@ public class BrokerServer {
     private final RaftNode raftNode;       
     private final List<RaftPeer> raftPeers; 
     private final BrokerMetrics metrics;
+    private TelemetryWebSocketServer telemetryServer;
 
     private volatile boolean running = false;
 
@@ -52,8 +53,8 @@ public class BrokerServer {
 
     public BrokerServer(BrokerConfig config) throws IOException {
         this.config = config;
-        this.logManager = new LogManager(config.getDataDir());
-        this.messageStore = new MessageStore(logManager);
+        this.logManager = new LogManager(config);
+        this.messageStore = new MessageStore(logManager, config);
         this.offsetManager = new OffsetManager(config.getDataDir());
         this.raftPeers = new ArrayList<>();
         this.metrics = BrokerMetrics.init(config);
@@ -65,7 +66,8 @@ public BrokerServer(BrokerConfig config) throws IOException {
                     config.getPeers(),
                     messageStore,
                     offsetManager,
-                    Paths.get(config.getDataDir())
+                    Paths.get(config.getDataDir()),
+                    config.getRaftCompactThreshold()
             );
 
             for (BrokerConfig.PeerAddress peer : config.getPeers()) {
@@ -74,6 +76,7 @@ public BrokerServer(BrokerConfig config) throws IOException {
                 raftNode.registerVoteHandler(peer.id(), raftPeer::sendRequestVote);
                 raftNode.registerAppendHandler(peer.id(), raftPeer::sendAppendEntries);
                 raftNode.registerPreVoteHandler(peer.id(), raftPeer::sendPreVote);
+                raftNode.registerInstallSnapshotHandler(peer.id(), raftPeer::sendInstallSnapshot);
             }
 
             logger.info("Cluster mode: nodeId={}, peers={}", config.getNodeId(), config.getPeers());
@@ -100,6 +103,10 @@ public BrokerServer() throws IOException {
         this(DEFAULT_PORT, DEFAULT_THREAD_POOL_SIZE);
     }
 
+    public int getActiveChannelsCount() {
+        return activeChannels != null ? activeChannels.size() : 0;
+    }
+
     public void start() throws IOException {
         try {
             messageStore.recover();
@@ -124,7 +131,7 @@ public void start() throws IOException {
                  @Override
                  public void initChannel(SocketChannel ch) {
                      ChannelPipeline p = ch.pipeline();
-                     p.addLast(new LengthFieldBasedFrameDecoder(10 * 1024 * 1024, 0, 4, 0, 4));
+                     p.addLast(new LengthFieldBasedFrameDecoder(256 * 1024 * 1024, 0, 4, 0, 4));
                      p.addLast(new ByteArrayDecoder());
                      p.addLast(new LengthFieldPrepender(4));
                      p.addLast(new ByteArrayEncoder());
@@ -139,6 +146,10 @@ public void initChannel(SocketChannel ch) {
             if (raftNode != null) {
                 raftNode.start();
             }
+            int wsPort = config.getPort() + 200;
+            telemetryServer = new TelemetryWebSocketServer(wsPort, this);
+            telemetryServer.start();
+
             logger.info("DRMQ Broker started on port {} with data directory {}",
                     config.getPort(), config.getDataDir());
 
@@ -179,6 +190,9 @@ public void shutdown() {
         if (raftNode != null) {
             raftNode.stop();
         }
+        if (telemetryServer != null) {
+            telemetryServer.shutdown();
+        }
 
         if (activeChannels != null) {
             activeChannels.close().awaitUninterruptibly();

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

@@ -98,6 +98,7 @@ private MessageEnvelope handleMessage(MessageEnvelope envelope) throws IOExcepti
             case REQUEST_VOTE_REQUEST -> handleRequestVoteRequest(envelope);
             case PRE_VOTE_REQUEST -> handlePreVoteRequest(envelope);
             case APPEND_ENTRIES_REQUEST -> handleAppendEntriesRequest(envelope);
+            case INSTALL_SNAPSHOT_REQUEST -> handleInstallSnapshotRequest(envelope);
             default -> createErrorResponse("Unknown message type: " + envelope.getType());
         };
     }
@@ -362,6 +363,7 @@ private MessageEnvelope createErrorResponse(String errorMessage, MessageType mes
             case REQUEST_VOTE_RESPONSE -> createRequestVoteErrorResponse();
             case PRE_VOTE_RESPONSE -> createPreVoteErrorResponse();
             case APPEND_ENTRIES_RESPONSE -> createAppendEntriesErrorResponse();
+            case INSTALL_SNAPSHOT_RESPONSE -> createInstallSnapshotErrorResponse();
             case PRODUCE_RESPONSE -> createProduceErrorResponse(errorMessage);
             case CONSUME_RESPONSE -> createConsumeErrorResponse(errorMessage);
             case COMMIT_OFFSET_RESPONSE -> createCommitOffsetErrorResponse(errorMessage);
@@ -395,6 +397,17 @@ private MessageEnvelope createAppendEntriesErrorResponse() {
                 .build();
     }
 
+    private MessageEnvelope createInstallSnapshotErrorResponse() {
+        InstallSnapshotResponse response = InstallSnapshotResponse.newBuilder()
+                .setTerm(0)
+                .build();
+
+        return MessageEnvelope.newBuilder()
+                .setType(MessageType.INSTALL_SNAPSHOT_RESPONSE)
+                .setPayload(response.toByteString())
+                .build();
+    }
+
     private MessageEnvelope createPreVoteErrorResponse() {
         PreVoteResponse response = PreVoteResponse.newBuilder()
                 .setTerm(0)
@@ -550,6 +563,51 @@ private MessageEnvelope handleAppendEntriesRequest(MessageEnvelope envelope) thr
         }
     }
 
+    private MessageEnvelope handleInstallSnapshotRequest(MessageEnvelope envelope) throws IOException {
+        long startNanos = System.nanoTime();
+        if (raftNode == null) {
+            BrokerMetrics.get().recordRaftRpc("install_snapshot", false,
+                    System.nanoTime() - startNanos);
+            return createInstallSnapshotErrorResponse();
+        }
+
+        InstallSnapshotRequest request = InstallSnapshotRequest.parseFrom(envelope.getPayload());
+        Future<InstallSnapshotResponse> future = null;
+        try {
+            future = rpcExecutor.submit(() -> raftNode.handleInstallSnapshot(request));
+            InstallSnapshotResponse response = future.get(10, TimeUnit.SECONDS);
+
+            BrokerMetrics.get().recordRaftRpc("install_snapshot", true,
+                    System.nanoTime() - startNanos);
+
+            return MessageEnvelope.newBuilder()
+                    .setType(MessageType.INSTALL_SNAPSHOT_RESPONSE)
+                    .setPayload(response.toByteString())
+                    .build();
+        } catch (TimeoutException e) {
+            logger.error("InstallSnapshot handler timed out");
+            BrokerMetrics.get().recordRaftRpc("install_snapshot", false,
+                    System.nanoTime() - startNanos);
+            if (future != null) {
+                future.cancel(true);
+            }
+            return createInstallSnapshotErrorResponse();
+        } catch (RejectedExecutionException e) {
+            logger.error("InstallSnapshot handler rejected: {}", e.getMessage());
+            BrokerMetrics.get().recordRaftRpc("install_snapshot", false,
+                    System.nanoTime() - startNanos);
+            return createInstallSnapshotErrorResponse();
+        } catch (Exception e) {
+            logger.error("InstallSnapshot handler failed: {}", e.getMessage());
+            BrokerMetrics.get().recordRaftRpc("install_snapshot", false,
+                    System.nanoTime() - startNanos);
+            if (future != null) {
+                future.cancel(true);
+            }
+            return createInstallSnapshotErrorResponse();
+        }
+    }
+
     private static long estimatePayloadBytes(java.util.List<StoredMessage> messages) {
         long total = 0;
         for (StoredMessage message : messages) {