github-mirror/CoreProtect
1
0 Fork 1
mirror of https://github.com/PlayPro/CoreProtect.git synced 2026-10-28 09:44:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reduce ClickHouse consumer payload retention

Browse Source
This commit is contained in:
Intelli committed 2026-10-04 11:05:16 -06:00
1 parent af78ea00d4
commit 1490dc7c89
2 files changed
+18 -5
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/main/java/net/coreprotect/consumer/process/Process.java
+15 -5
View File
@@ -268,6 +268,7 @@ public class Process {
entitySpawnUpdates = hasEntitySpawnUpdates ? writeBatch.entitySpawnUpdates() : null;
if (entitySpawnUpdates != null) {
entitySpawnUpdates.prefetch(entitySpawnUpdateData);
entitySpawnUpdateData.clear();
}
processingStarted = true;
for (int i = 0; i < consumerDataSize; i++) {
@@ -527,6 +528,7 @@ public class Process {
boolean interrupted = Consumer.interrupt;
boolean batchLimitReached = writeBatch.shouldCommit();
if ((interrupted || batchLimitReached) && !isolatedTransaction) {
int batchStart = processedThrough;
TransactionOutcome outcome = commit(writeBatch);
if (outcome == TransactionOutcome.COMMITTED) {
processedThrough = i + 1;
@@ -536,6 +538,9 @@ public class Process {
completeFailedConsumerBatch(processId, consumerData, users, consumerObject, processedThrough, i + 1, outcome == TransactionOutcome.RETAINED);
return;
}
if (ConfigHandler.databaseType.isClickHouse()) {
releaseProcessedConsumerData(processId, consumerData, users, consumerObject, batchStart, processedThrough);
}
boolean backlog = batchLimitReached && ConfigHandler.databaseType.isClickHouse() && i + 1 < consumerDataSize;
if (interrupted || backlog) {
try {
@@ -938,16 +943,21 @@ public class Process {
private static void discardProcessedConsumerData(int processId, ArrayList<Object[]> consumerData, Map<Integer, String[]> users, Map<Integer, Object> consumerObject, int count) {
int processed = Math.min(count, consumerData.size());
for (int index = 0; index < processed; index++) {
releaseProcessedConsumerData(processId, consumerData, users, consumerObject, 0, processed);
consumerData.subList(0, processed).clear();
}
private static void releaseProcessedConsumerData(int processId, ArrayList<Object[]> consumerData, Map<Integer, String[]> users, Map<Integer, Object> consumerObject, int fromIndex, int toIndex) {
for (int index = fromIndex; index < toIndex; index++) {
Object[] data = consumerData.get(index);
preparationFailures.remove(data);
if (data == null) {
continue;
}
int id = (int) data[0];
preparationFailures.remove(data);
Integer id = (Integer) data[0];
discardConsumerData(processId, users, consumerObject, id, (int) data[1]);
consumerData.set(index, null);
}
consumerData.subList(0, processed).clear();
}
private static void discardFailedConsumerData(int processId, ArrayList<Object[]> consumerData, Map<Integer, String[]> users,
@@ -1040,7 +1050,7 @@ public class Process {
return false;
}
private static void discardConsumerData(int processId, Map<Integer, String[]> users, Map<Integer, Object> consumerObject, int id, int action) {
private static void discardConsumerData(int processId, Map<Integer, String[]> users, Map<Integer, Object> consumerObject, Integer id, int action) {
Object object = consumerObject.get(id);
if (action == -1) {
if (object instanceof EntityContainerRollbackUpdate) {
src/main/java/net/coreprotect/database/clickhouse/ClickHouseRowBinaryBuffer.java
+3
View File
@@ -7,6 +7,7 @@ import java.io.InputStream;
import java.sql.SQLException;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -109,6 +110,7 @@ final class ClickHouseRowBinaryBuffer implements AutoCloseable {
if (rowStarted) {
throw new IllegalStateException("ClickHouse RowBinary row is incomplete");
}
Arrays.fill(values, null);
sealed = true;
}
@@ -161,6 +163,7 @@ final class ClickHouseRowBinaryBuffer implements AutoCloseable {
sealed = true;
rows.reset();
rowSpans.clear();
Arrays.fill(values, null);
}
private void ensureWritable() {