Skip to content

[Subscription] Bound consensus prefetch memory usage #18298

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[Subscription] Bound consensus prefetch memory usage #18298

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
010635c
[Subscription] Bound consensus prefetch memory usage
Caideyipi Jul 24, 2026
d41d8bc
[Subscription] Accept fenced consensus commits as no-ops
Caideyipi Jul 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
124 changes: 84 additions & 40 deletions iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,13 @@
public class DataNodeMemoryConfig {
private static final Logger LOGGER = LoggerFactory.getLogger(DataNodeMemoryConfig.class);

private static final int LEGACY_QUERY_MEMORY_COMPONENT_COUNT = 8;
private static final int QUERY_MEMORY_COMPONENT_COUNT = 9;
private static final int SUBSCRIPTION_MEMORY_INDEX = 8;
private static final int[] DEFAULT_QUERY_MEMORY_PROPORTIONS = {
1, 100, 200, 50, 200, 200, 200, 50, 250
};

public static final String SCHEMA_CACHE = "SchemaCache";
public static final String SCHEMA_REGION = "SchemaRegion";
public static final String PARTITION_CACHE = "PartitionCache";
Expand Down Expand Up @@ -131,6 +138,9 @@ public class DataNodeMemoryConfig {
/** Memory manager proportion for timeIndex */
private MemoryManager timeIndexMemoryManager;

/** Memory manager for subscription */
private MemoryManager subscriptionMemoryManager;

/** Memory manager for the mtree */
private MemoryManager schemaEngineMemoryManager;

Expand Down Expand Up @@ -464,6 +474,42 @@ private void initCompactionMemoryManager(long compactionMemorySize) {
"Compaction", compactionMemorySize - fixedMemoryCost);
}

static int[] resolveQueryMemoryProportions(
final String configuredProportions, final boolean subscriptionEnabled) {
final int[] resolvedProportions = DEFAULT_QUERY_MEMORY_PROPORTIONS.clone();
if (configuredProportions != null) {
final String[] proportions = configuredProportions.split(":");
if (proportions.length != LEGACY_QUERY_MEMORY_COMPONENT_COUNT
&& proportions.length != QUERY_MEMORY_COMPONENT_COUNT) {
throw new IllegalArgumentException();
}
for (int i = 0; i < proportions.length; i++) {
resolvedProportions[i] = Integer.parseInt(proportions[i].trim());
if (resolvedProportions[i] < 0) {
throw new IllegalArgumentException();
}
}
if (proportions.length == LEGACY_QUERY_MEMORY_COMPONENT_COUNT) {
int legacyProportionSum = 0;
for (int i = 0; i < LEGACY_QUERY_MEMORY_COMPONENT_COUNT; i++) {
legacyProportionSum += resolvedProportions[i];
}
resolvedProportions[SUBSCRIPTION_MEMORY_INDEX] = legacyProportionSum / 4;
}
}
if (!subscriptionEnabled) {
resolvedProportions[SUBSCRIPTION_MEMORY_INDEX] = 0;
}
int proportionSum = 0;
for (final int proportion : resolvedProportions) {
proportionSum += proportion;
}
if (proportionSum <= 0) {
throw new IllegalArgumentException();
}
return resolvedProportions;
}

@SuppressWarnings("squid:S3518")
private void initQueryEngineMemoryAllocate(
MemoryManager queryEngineMemoryManager, TrimProperties properties) {
Expand Down Expand Up @@ -494,50 +540,41 @@ private void initQueryEngineMemoryAllocate(
LOGGER.error(String.format(DataNodeMiscMessages.FAIL_RELOAD_CONFIGURATION_FMT, e));
}

long maxMemoryAvailable = queryEngineMemoryManager.getTotalMemorySizeInBytes();
String queryMemoryAllocateProportion =
properties.getProperty("chunk_timeseriesmeta_free_memory_proportion");
long maxMemoryAvailable = queryEngineMemoryManager.getTotalMemorySizeInBytes();

long bloomFilterCacheMemorySize = maxMemoryAvailable / 1001;
long chunkCacheMemorySize = maxMemoryAvailable * 100 / 1001;
long timeSeriesMetaDataCacheMemorySize = maxMemoryAvailable * 200 / 1001;
long coordinatorMemorySize = maxMemoryAvailable * 50 / 1001;
long operatorsMemorySize = maxMemoryAvailable * 200 / 1001;
long dataExchangeMemorySize = maxMemoryAvailable * 200 / 1001;
long timeIndexMemorySize = maxMemoryAvailable * 200 / 1001;
if (queryMemoryAllocateProportion != null) {
String[] proportions = queryMemoryAllocateProportion.split(":");
int proportionSum = 0;
for (String proportion : proportions) {
proportionSum += Integer.parseInt(proportion.trim());
}
if (proportionSum != 0) {
try {
bloomFilterCacheMemorySize =
maxMemoryAvailable * Integer.parseInt(proportions[0].trim()) / proportionSum;
chunkCacheMemorySize =
maxMemoryAvailable * Integer.parseInt(proportions[1].trim()) / proportionSum;
timeSeriesMetaDataCacheMemorySize =
maxMemoryAvailable * Integer.parseInt(proportions[2].trim()) / proportionSum;
coordinatorMemorySize =
maxMemoryAvailable * Integer.parseInt(proportions[3].trim()) / proportionSum;
operatorsMemorySize =
maxMemoryAvailable * Integer.parseInt(proportions[4].trim()) / proportionSum;
dataExchangeMemorySize =
maxMemoryAvailable * Integer.parseInt(proportions[5].trim()) / proportionSum;
timeIndexMemorySize =
maxMemoryAvailable * Integer.parseInt(proportions[6].trim()) / proportionSum;
} catch (Exception e) {
throw new IllegalArgumentException(
String.format(
DataNodeMiscMessages
.MISC_EXCEPTION_EACH_SUBSECTION_OF_CONFIGURATION_ITEM_CHUNKMETA_CHUNK_TIMESERIESMETA_77A43CE2,
queryMemoryAllocateProportion),
e);
}
}
boolean subscriptionEnabled =
Boolean.parseBoolean(
properties.getProperty("subscription_enabled", Boolean.TRUE.toString()));
final int[] queryMemoryProportions;
try {
queryMemoryProportions =
resolveQueryMemoryProportions(queryMemoryAllocateProportion, subscriptionEnabled);
} catch (Exception e) {
throw new IllegalArgumentException(
String.format(
DataNodeMiscMessages
.MISC_EXCEPTION_EACH_SUBSECTION_OF_CONFIGURATION_ITEM_CHUNKMETA_CHUNK_TIMESERIESMETA_77A43CE2,
queryMemoryAllocateProportion),
e);
}
int proportionSum = 0;
for (int proportion : queryMemoryProportions) {
proportionSum += proportion;
}

long bloomFilterCacheMemorySize =
maxMemoryAvailable * queryMemoryProportions[0] / proportionSum;
long chunkCacheMemorySize = maxMemoryAvailable * queryMemoryProportions[1] / proportionSum;
long timeSeriesMetaDataCacheMemorySize =
maxMemoryAvailable * queryMemoryProportions[2] / proportionSum;
long coordinatorMemorySize = maxMemoryAvailable * queryMemoryProportions[3] / proportionSum;
long operatorsMemorySize = maxMemoryAvailable * queryMemoryProportions[4] / proportionSum;
long dataExchangeMemorySize = maxMemoryAvailable * queryMemoryProportions[5] / proportionSum;
long timeIndexMemorySize = maxMemoryAvailable * queryMemoryProportions[6] / proportionSum;
long subscriptionMemorySize =
maxMemoryAvailable * queryMemoryProportions[SUBSCRIPTION_MEMORY_INDEX] / proportionSum;

// metadata cache is disabled, we need to move all their allocated memory to other parts
if (!isMetaDataCacheEnable()) {
long sum =
Expand Down Expand Up @@ -567,6 +604,9 @@ private void initQueryEngineMemoryAllocate(
queryEngineMemoryManager.getOrCreateMemoryManager("DataExchange", dataExchangeMemorySize);
timeIndexMemoryManager =
queryEngineMemoryManager.getOrCreateMemoryManager("TimeIndex", timeIndexMemorySize);
subscriptionMemoryManager =
queryEngineMemoryManager.getOrCreateMemoryManager(
"Subscription", subscriptionMemorySize, true);

// must be called after dataExchangeMemoryManager being inited.
setQueryThreadCount(
Expand Down Expand Up @@ -730,6 +770,10 @@ public MemoryManager getTimeIndexMemoryManager() {
return timeIndexMemoryManager;
}

public MemoryManager getSubscriptionMemoryManager() {
return subscriptionMemoryManager;
}

public MemoryManager getSchemaEngineMemoryManager() {
return schemaEngineMemoryManager;
}
Expand Down
26 changes: 25 additions & 1 deletion ...de/src/main/java/org/apache/iotdb/db/service/metrics/memory/QueryEngineMemoryMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ public class QueryEngineMemoryMetrics implements IMetricSet {
private static final String QUERY_ENGINE_DATA_EXCHANGE = "QueryEngine-DataExchange";
private static final String QUERY_ENGINE_TIME_INDEX = "QueryEngine-TimeIndex";
private static final String QUERY_ENGINE_COORDINATOR = "QueryEngine-Coordinator";
private static final String QUERY_ENGINE_SUBSCRIPTION = "QueryEngine-Subscription";

@Override
public void bindTo(AbstractMetricService metricService) {
Expand Down Expand Up @@ -211,6 +212,28 @@ public void bindTo(AbstractMetricService metricService) {
GlobalMemoryMetrics.ON_HEAP,
Tag.LEVEL.toString(),
GlobalMemoryMetrics.LEVELS[2]);
metricService.createAutoGauge(
Metric.MEMORY_THRESHOLD_SIZE.toString(),
MetricLevel.IMPORTANT,
memoryConfig.getSubscriptionMemoryManager(),
MemoryManager::getTotalMemorySizeInBytes,
Tag.NAME.toString(),
QUERY_ENGINE_SUBSCRIPTION,
Tag.TYPE.toString(),
GlobalMemoryMetrics.ON_HEAP,
Tag.LEVEL.toString(),
GlobalMemoryMetrics.LEVELS[2]);
metricService.createAutoGauge(
Metric.MEMORY_ACTUAL_SIZE.toString(),
MetricLevel.IMPORTANT,
memoryConfig.getSubscriptionMemoryManager(),
MemoryManager::getUsedMemorySizeInBytes,
Tag.NAME.toString(),
QUERY_ENGINE_SUBSCRIPTION,
Tag.TYPE.toString(),
GlobalMemoryMetrics.ON_HEAP,
Tag.LEVEL.toString(),
GlobalMemoryMetrics.LEVELS[2]);
}

@Override
Expand All @@ -231,7 +254,8 @@ public void unbindFrom(AbstractMetricService metricService) {
QUERY_ENGINE_OPERATORS,
QUERY_ENGINE_DATA_EXCHANGE,
QUERY_ENGINE_TIME_INDEX,
QUERY_ENGINE_COORDINATOR)
QUERY_ENGINE_COORDINATOR,
QUERY_ENGINE_SUBSCRIPTION)
.forEach(
name -> {
metricService.remove(
Expand Down
21 changes: 14 additions & 7 deletions ...de/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -249,15 +249,22 @@ public List<SubscriptionCommitContext> commit(
getQueueForCommitContext(queues, commitContext);
boolean handled = false;
if (Objects.nonNull(assignedQueue)) {
final boolean success;
if (!nack) {
success = assignedQueue.ackSilent(consumerId, commitContext);
} else {
success = assignedQueue.nackSilent(consumerId, commitContext);
}
if (success) {
if (assignedQueue.isCommitContextOutdated(commitContext)) {
// A seek or reboot has already fenced this context. Accept the delayed commit as a no-op
// so the client does not retry it, while keeping the current commit frontier unchanged.
successfulCommitContexts.add(commitContext);
handled = true;
} else {
final boolean success;
if (!nack) {
success = assignedQueue.ackSilent(consumerId, commitContext);
} else {
success = assignedQueue.nackSilent(consumerId, commitContext);
}
if (success) {
successfulCommitContexts.add(commitContext);
handled = true;
}
}
}
if (!handled) {
Expand Down
Loading
Loading