[ISSUE #10527] Reduce auxiliary component allocation via AtomicLong, string caches, StringBuilder reuse, and dirty flag

store/src/main/java/org/apache/rocketmq/store/AppendMessageResult.java

@@ -84,6 +84,18 @@ public AppendMessageResult(AppendMessageStatus status, long wroteOffset, int wro
         this.msgNum = msgNum;
     }
 
+    public AppendMessageResult(AppendMessageStatus status, long wroteOffset, int wroteBytes, String msgId,
+            long storeTimestamp, long logicsOffset, long pagecacheRT, int msgNum) {
+        this.status = status;
+        this.wroteOffset = wroteOffset;
+        this.wroteBytes = wroteBytes;
+        this.msgId = msgId;
+        this.storeTimestamp = storeTimestamp;
+        this.logicsOffset = logicsOffset;
+        this.pagecacheRT = pagecacheRT;
+        this.msgNum = msgNum;
+    }
+
     public long getPagecacheRT() {
         return pagecacheRT;
     }

store/src/main/java/org/apache/rocketmq/store/index/IndexService.java

@@ -49,6 +49,7 @@ public class IndexService implements CommitLogDispatchStore {
     private final String storePath;
     private final ArrayList<IndexFile> indexFileList = new ArrayList<>();
     private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
+    private static final ThreadLocal<StringBuilder> KEY_BUILDER = ThreadLocal.withInitial(() -> new StringBuilder(128));
 
     public IndexService(final DefaultMessageStore store) {
         this.defaultMessageStore = store;
@@ -215,10 +216,15 @@ public QueryOffsetResult queryOffset(String topic, String key, int maxNum, long
     }
 
     private String buildKey(final String topic, final String key) {
-        return topic + "#" + key;
+        StringBuilder keyBuilder = KEY_BUILDER.get();
+        keyBuilder.setLength(0);
+        return keyBuilder.append(topic).append('#').append(key).toString();
     }
+
     private String buildKey(final String topic, final String key, final String indexType) {
-        return topic + "#" + indexType + "#" + key;
+        StringBuilder keyBuilder = KEY_BUILDER.get();
+        keyBuilder.setLength(0);
+        return keyBuilder.append(topic).append('#').append(indexType).append('#').append(key).toString();
     }
 
     public void buildIndex(DispatchRequest req) {

store/src/main/java/org/apache/rocketmq/store/timer/TimerWheel.java

@@ -55,6 +55,8 @@ protected ByteBuffer initialValue() {
     };
     private final int wheelLength;
 
+    private volatile boolean dirty;
+
     private long snapOffset;
 
     public TimerWheel(String fileName, int slotsTotal, int precisionMs) throws IOException {
@@ -128,14 +130,26 @@ public void flush() {
         if (mappedByteBuffer == null) {
             return;
         }
+        if (!dirty) {
+            return;
+        }
+        // Clear the flag before diffing: a concurrent putSlot during the diff re-marks it and the
+        // next flush picks the change up, so an interleaving can only cause an extra flush, never a
+        // missed one. All threads share the same underlying direct buffer (localBuffer duplicates
+        // only isolate position/limit), so the data itself is always visible here.
+        dirty = false;
         ByteBuffer bf = localBuffer.get();
-        bf.position(0);
-        bf.limit(wheelLength);
-        mappedByteBuffer.position(0);
-        mappedByteBuffer.limit(wheelLength);
-        for (int i = 0; i < wheelLength; i++) {
-            if (bf.get(i) != mappedByteBuffer.get(i)) {
-                mappedByteBuffer.put(i, bf.get(i));
+        int longAligned = wheelLength & ~7;
+        for (int i = 0; i < longAligned; i += 8) {
+            long local = bf.getLong(i);
+            if (local != mappedByteBuffer.getLong(i)) {
+                mappedByteBuffer.putLong(i, local);
+            }
+        }
+        for (int i = longAligned; i < wheelLength; i++) {
+            byte b = bf.get(i);
+            if (b != mappedByteBuffer.get(i)) {
+                mappedByteBuffer.put(i, b);
             }
         }
         this.mappedByteBuffer.force();
@@ -287,11 +301,10 @@ public int getSlotIndex(long timeMs) {
 
     public void putSlot(long timeMs, long firstPos, long lastPos) {
         localBuffer.get().position(getSlotIndex(timeMs) * Slot.SIZE);
-        // To be compatible with previous version.
-        // The previous version's precision is fixed at 1000ms and it store timeMs / 1000 in slot.
         localBuffer.get().putLong(timeMs / precisionMs);
         localBuffer.get().putLong(firstPos);
         localBuffer.get().putLong(lastPos);
+        dirty = true;
     }
 
     public void putSlot(long timeMs, long firstPos, long lastPos, int num, int magic) {
@@ -301,6 +314,7 @@ public void putSlot(long timeMs, long firstPos, long lastPos, int num, int magic
         localBuffer.get().putLong(lastPos);
         localBuffer.get().putInt(num);
         localBuffer.get().putInt(magic);
+        dirty = true;
     }
 
     public void reviseSlot(long timeMs, long firstPos, long lastPos, boolean force) {
@@ -313,11 +327,13 @@ public void reviseSlot(long timeMs, long firstPos, long lastPos, boolean force)
         } else {
             if (IGNORE != firstPos) {
                 localBuffer.get().putLong(firstPos);
+                dirty = true;
             } else {
                 localBuffer.get().getLong();
             }
             if (IGNORE != lastPos) {
                 localBuffer.get().putLong(lastPos);
+                dirty = true;
             }
         }
     }

store/src/test/java/org/apache/rocketmq/store/stats/BrokerStatsManagerTest.java

@@ -45,7 +45,7 @@ public class BrokerStatsManagerTest {
     private BrokerStatsManager brokerStatsManager;
 
     private static final String TOPIC = "TOPIC_TEST";
-    private static final Integer QUEUE_ID = 0;
+    private static final int QUEUE_ID = 0;
     private static final String GROUP_NAME = "GROUP_TEST";
     private static final String CLUSTER_NAME = "DefaultCluster";