荣耀配置

src/main/java/com/adx/tencent/honor/service/HonorColdWorker.java

@@ -55,6 +55,7 @@ public class HonorColdWorker {
         }
         if (failed != null) throw failed;
         hotStore.ack(group, acked);
+        hotStore.delete(acked);
         if (streamMaxLen > 0) hotStore.trim(streamMaxLen);
         return events.size();
     }

src/main/java/com/adx/tencent/honor/store/HonorHotStore.java

@@ -240,6 +240,11 @@ public class HonorHotStore {
         redis.opsForStream().acknowledge(stream, group, ids.toArray(String[]::new));
     }
 
+    public void delete(List<String> ids) {
+        if (ids == null || ids.isEmpty()) return;
+        redis.opsForStream().delete(stream, ids.toArray(String[]::new));
+    }
+
     public void trim(long maxLen) {
         if (maxLen > 0) redis.opsForStream().trim(stream, maxLen, false);
     }

src/main/java/com/adx/tencent/kuaishou/service/KuaishouColdWorker.java

@@ -55,6 +55,7 @@ public class KuaishouColdWorker {
         }
         if (failed != null) throw failed;
         hotStore.ack(group, acked);
+        hotStore.delete(acked);
         if (streamMaxLen > 0) hotStore.trim(streamMaxLen);
         return events.size();
     }

src/main/java/com/adx/tencent/kuaishou/store/KuaishouHotStore.java

@@ -282,6 +282,11 @@ public class KuaishouHotStore {
         redis.opsForStream().acknowledge(stream, group, ids.toArray(String[]::new));
     }
 
+    public void delete(List<String> ids) {
+        if (ids == null || ids.isEmpty()) return;
+        redis.opsForStream().delete(stream, ids.toArray(String[]::new));
+    }
+
     public void trim(long maxLen) {
         if (maxLen > 0) redis.opsForStream().trim(stream, maxLen, false);
     }

src/main/java/com/adx/tencent/storage/RedisHotStore.java

@@ -377,6 +377,11 @@ public class RedisHotStore {
         redis.opsForStream().acknowledge(stream, group, idArr);
     }
 
+    public void delete(List<String> ids) {
+        if (ids == null || ids.isEmpty()) return;
+        redis.opsForStream().delete(stream, ids.toArray(String[]::new));
+    }
+
     /**
      * 对应 Go Trim：XTRIMAPPROX。
      */

src/main/java/com/adx/tencent/vivo/service/VivoColdWorker.java

@@ -55,6 +55,7 @@ public class VivoColdWorker {
         }
         if (failed != null) throw failed;
         hotStore.ack(group, acked);
+        hotStore.delete(acked);
         if (streamMaxLen > 0) hotStore.trim(streamMaxLen);
         return events.size();
     }

src/main/java/com/adx/tencent/vivo/store/VivoHotStore.java

@@ -282,6 +282,11 @@ public class VivoHotStore {
         redis.opsForStream().acknowledge(stream, group, ids.toArray(String[]::new));
     }
 
+    public void delete(List<String> ids) {
+        if (ids == null || ids.isEmpty()) return;
+        redis.opsForStream().delete(stream, ids.toArray(String[]::new));
+    }
+
     public void trim(long maxLen) {
         if (maxLen > 0) redis.opsForStream().trim(stream, maxLen, false);
     }

src/main/java/com/adx/tencent/worker/ColdWorker.java

@@ -89,6 +89,7 @@ public class ColdWorker {
         }
 
         hotStore.ack(group, acked);
+        hotStore.delete(acked);
         if (streamMaxLen > 0) {
             hotStore.trim(streamMaxLen);
         }