wuchong commented on code in PR #3614:
URL: https://github.com/apache/fluss/pull/3614#discussion_r3570748264
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -577,6 +588,30 @@ public class ConfigOptions {
"The rack for the tabletServer. This will be used
in rack aware bucket assignment "
+ "for fault tolerance. Examples: `RACK1`,
`cn-hangzhou-server10`");
+ public static final ConfigOption<Double>
TABLET_SERVER_ADVERTISED_RESOURCE_CPU_CORES =
+ key("tablet-server.advertised-resource.cpu-cores")
+ .doubleType()
+ .noDefaultValue()
+ .withDeprecatedKeys("tablet-server.resource.cpu-cores")
+ .withDescription(
+ "The CPU capacity, in cores, that this tablet
server advertises to the "
+ + "CoordinatorServer for resource
reporting. This option does not limit "
+ + "CPU usage or configure a cgroup CPU
quota. If not configured, the tablet "
+ + "server detects the value from cgroup
CPU quota or the JVM runtime.");
+
+ public static final ConfigOption<MemorySize>
TABLET_SERVER_ADVERTISED_RESOURCE_MEMORY_SIZE =
+ key("tablet-server.advertised-resource.memory-size")
+ .memoryType()
+ .noDefaultValue()
+ .withDeprecatedKeys("tablet-server.resource.memory-size")
Review Comment:
ditto
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java:
##########
@@ -715,40 +734,50 @@ public CompletableFuture<CreatePartitionResponse>
createPartition(
authorizeTable(OperationType.WRITE, tablePath);
CreatePartitionResponse response = new CreatePartitionResponse();
- TableRegistration table =
metadataManager.getTableRegistration(tablePath);
- if (!table.isPartitioned()) {
+ TableInfo tableInfo = metadataManager.getTable(tablePath);
+ if (!tableInfo.isPartitioned()) {
throw new TableNotPartitionedException(
"Only partitioned table support create partition.");
}
// first, validate the partition spec, and get resolved partition spec.
PartitionSpec partitionSpec =
getPartitionSpec(request.getPartitionSpec());
- validatePartitionSpec(tablePath, table.partitionKeys, partitionSpec,
true);
+ validatePartitionSpec(tablePath, tableInfo.getPartitionKeys(),
partitionSpec, true);
// second, check whether the partition is out-of-date.
validateAutoPartitionTime(
partitionSpec,
- table.partitionKeys,
- table.getTableConfig().getAutoPartitionStrategy());
+ tableInfo.getPartitionKeys(),
+ tableInfo.getTableConfig().getAutoPartitionStrategy());
ResolvedPartitionSpec partitionToCreate =
- ResolvedPartitionSpec.fromPartitionSpec(table.partitionKeys,
partitionSpec);
+ ResolvedPartitionSpec.fromPartitionSpec(
+ tableInfo.getPartitionKeys(), partitionSpec);
+ if (request.isIgnoreIfNotExists()
+ && metadataManager
+ .getPartitions(tablePath)
+ .contains(partitionToCreate.getPartitionName())) {
Review Comment:
Use
`org.apache.fluss.server.coordinator.MetadataManager.getOptionalPartitionRegistration`
to check partition exists which is more cheap?
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -577,6 +588,30 @@ public class ConfigOptions {
"The rack for the tabletServer. This will be used
in rack aware bucket assignment "
+ "for fault tolerance. Examples: `RACK1`,
`cn-hangzhou-server10`");
+ public static final ConfigOption<Double>
TABLET_SERVER_ADVERTISED_RESOURCE_CPU_CORES =
+ key("tablet-server.advertised-resource.cpu-cores")
+ .doubleType()
+ .noDefaultValue()
+ .withDeprecatedKeys("tablet-server.resource.cpu-cores")
Review Comment:
We don't need to keep compatibility with `tablet-server.resource.cpu-cores`,
as it is never exposed to users.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/ReplicaCapacityController.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.coordinator;
+
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.cluster.ServerReconfigurable;
+import org.apache.fluss.exception.ConfigException;
+import org.apache.fluss.exception.InsufficientKvLeaderReplicaCapacityException;
+import org.apache.fluss.metrics.MetricNames;
+import org.apache.fluss.server.metadata.CoordinatorMetadataCache;
+import org.apache.fluss.server.metadata.ServerInfo;
+import org.apache.fluss.server.metrics.group.CoordinatorMetricGroup;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Set;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Controls replica creation admission based on in-memory cluster capacity. */
+public class ReplicaCapacityController implements ServerReconfigurable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ReplicaCapacityController.class);
+
+ /** Negative capacity means the automatic capacity limit is disabled. */
+ public static final long CAPACITY_LIMIT_DISABLED = -1L;
+
+ private final Object lock = new Object();
+ private final CoordinatorMetadataCache metadataCache;
+
+ private long kvLeaderReplicaMemoryReservedBytes;
Review Comment:
can be volatile, then we don't need to use lock synchronized
##########
fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataCache.java:
##########
@@ -94,6 +94,19 @@ public Set<TabletServerInfo> getAliveTabletServerInfos() {
return Collections.unmodifiableSet(tabletServerInfos);
}
+ public Set<ServerInfo> getLiveTabletServerInfos() {
+ Map<Integer, ServerInfo> aliveTabletServers =
metadataSnapshot.aliveTabletServers;
Review Comment:
Get `metadataSnapshot` into a local variable first, then access it via the
local variable, otherwise, there is data consistency problem.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java:
##########
@@ -597,6 +604,14 @@ private void loadAssignment(
}
}
+ @VisibleForTesting
+ void trackKvBucketsForLoadedAssignment(long tableId, Set<TableBucket>
tableBuckets) {
+ TableInfo tableInfo = coordinatorContext.getTableInfoById(tableId);
+ if (tableInfo == null || tableInfo.hasPrimaryKey()) {
Review Comment:
Add comment to explain why tableInfo not found is treated as kv buckets.
##########
fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServerResourceProbeTest.java:
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.tablet;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.server.metadata.TabletServerResource;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link TabletServerResourceProbe}. */
+class TabletServerResourceProbeTest {
+
+ @TempDir private Path tempDir;
+
+ @Test
+ void testProbeFromExplicitConfig() {
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.TABLET_SERVER_ADVERTISED_RESOURCE_CPU_CORES,
8.0);
+ conf.set(ConfigOptions.TABLET_SERVER_ADVERTISED_RESOURCE_MEMORY_SIZE,
new MemorySize(1024));
+
+ TabletServerResource resource = new TabletServerResourceProbe(conf,
tempDir).probe();
+
+ assertThat(resource.getCpuCores()).isEqualTo(8.0);
+ assertThat(resource.getMemoryBytes()).isEqualTo(1024);
+ }
+
+ @Test
+ void testProbeFromDeprecatedResourceConfig() {
+ Configuration conf = new Configuration();
+ conf.setString("tablet-server.resource.cpu-cores", "4.0");
+ conf.setString("tablet-server.resource.memory-size", "2kb");
+
+ TabletServerResource resource = new TabletServerResourceProbe(conf,
tempDir).probe();
+
+ assertThat(resource.getCpuCores()).isEqualTo(4.0);
+ assertThat(resource.getMemoryBytes()).isEqualTo(2048);
Review Comment:
We don't need to keep compatible with them as they are never exposed to
users.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java:
##########
@@ -821,6 +842,7 @@ private void processCreateTable(CreateTableEvent
createTableEvent) {
coordinatorContext.putTableInfo(tableInfo);
TableAssignment tableAssignment =
createTableEvent.getTableAssignment();
tableManager.onCreateNewTable(tablePath, tableInfo.getTableId(),
tableAssignment);
+ updateObservedKvLeaderReplicaCount();
Review Comment:
**[P2] Centralize the observed KV bucket count update in `TableManager`**
Could we move `updateObservedKvLeaderReplicaCount()` into `TableManager` and
invoke it immediately after `CoordinatorContext#addKvBuckets` /
`removeKvBuckets`?
Currently, every `CoordinatorEventProcessor` caller must remember to refresh
the derived count after calling `TableManager#onCreateNewTable`,
`onCreateNewPartition`, `onDeleteTable`, or `onDeletePartition`. Missing any
current or future call site would leave the admission controller with a stale
count.
For example:
```java
private void updateObservedKvLeaderReplicaCount() {
replicaCapacityController.updateObservedKvLeaderReplicaCount(
coordinatorContext.getKvBucketCount());
}
private void onCreateNewTableBucket(long tableId, Set<TableBucket>
tableBuckets) {
TableInfo tableInfo = coordinatorContext.getTableInfoById(tableId);
if (tableInfo != null && tableInfo.hasPrimaryKey()) {
coordinatorContext.addKvBuckets(tableBuckets);
updateObservedKvLeaderReplicaCount();
}
// Continue with state-machine transitions...
}
private void onDeleteTableBucket(
Set<TableBucket> tableBuckets, Set<TableBucketReplica> allReplicas) {
// Remove unconditionally because orphan assignments may also have been
tracked as KV buckets.
coordinatorContext.removeKvBuckets(tableBuckets);
updateObservedKvLeaderReplicaCount();
// Continue with state-machine transitions...
}
```
Keeping the source-of-truth mutation and derived-count publication together
also ensures that the count remains consistent if a later state-machine
transition throws.
`CoordinatorEventProcessor` would then only need explicit synchronization at
the two lifecycle boundaries that bypass `TableManager`:
- after failover initialization finishes loading all assignments;
- after shutdown resets `CoordinatorContext`.
This would remove the duplicated updates from the normal create, delete, and
`ResumeDropEvent` paths and make future lifecycle changes much less error-prone.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/ReplicaCapacityController.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.coordinator;
+
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.cluster.ServerReconfigurable;
+import org.apache.fluss.exception.ConfigException;
+import org.apache.fluss.exception.InsufficientKvLeaderReplicaCapacityException;
+import org.apache.fluss.metrics.MetricNames;
+import org.apache.fluss.server.metadata.CoordinatorMetadataCache;
+import org.apache.fluss.server.metadata.ServerInfo;
+import org.apache.fluss.server.metrics.group.CoordinatorMetricGroup;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Set;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Controls replica creation admission based on in-memory cluster capacity. */
+public class ReplicaCapacityController implements ServerReconfigurable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ReplicaCapacityController.class);
+
+ /** Negative capacity means the automatic capacity limit is disabled. */
+ public static final long CAPACITY_LIMIT_DISABLED = -1L;
+
+ private final Object lock = new Object();
+ private final CoordinatorMetadataCache metadataCache;
+
+ private long kvLeaderReplicaMemoryReservedBytes;
+ private volatile long observedKvLeaderReplicaCount;
+
+ public ReplicaCapacityController(
+ Configuration conf,
+ CoordinatorMetadataCache metadataCache,
+ CoordinatorMetricGroup coordinatorMetricGroup) {
+ this(conf, metadataCache);
+ registerMetrics(
+ checkNotNull(coordinatorMetricGroup, "coordinatorMetricGroup
should not be null."));
+ }
+
+ @VisibleForTesting
+ ReplicaCapacityController(Configuration conf, CoordinatorMetadataCache
metadataCache) {
+ this.metadataCache = checkNotNull(metadataCache, "metadataCache should
not be null.");
+ long memoryReservedBytes = getKvLeaderReplicaMemoryReservedBytes(conf);
+ validateKvLeaderReplicaMemoryReservedBytes(memoryReservedBytes);
+ this.kvLeaderReplicaMemoryReservedBytes = memoryReservedBytes;
+ }
+
+ @Override
+ public void validate(Configuration newConfig) throws ConfigException {
+ long memoryReservedBytes =
getKvLeaderReplicaMemoryReservedBytes(newConfig);
+ validateKvLeaderReplicaMemoryReservedBytes(memoryReservedBytes);
+ }
+
+ @Override
+ public void reconfigure(Configuration newConfig) throws ConfigException {
+ long memoryReservedBytes =
getKvLeaderReplicaMemoryReservedBytes(newConfig);
+ synchronized (lock) {
+ kvLeaderReplicaMemoryReservedBytes = memoryReservedBytes;
+ }
+ }
+
+ /** Checks whether the requested KV leader replicas fit the current
observed capacity. */
+ public void checkCanCreateKvLeaderReplicas(long
requestedKvLeaderReplicaCount) {
+ if (requestedKvLeaderReplicaCount <= 0) {
+ return;
+ }
+
+ synchronized (lock) {
+ long capacity = getKvLeaderReplicaCapacityLocked();
+ long observedCount = observedKvLeaderReplicaCount;
+ if (capacity != CAPACITY_LIMIT_DISABLED
+ && (observedCount > capacity
+ || requestedKvLeaderReplicaCount > capacity -
observedCount)) {
+ throw new InsufficientKvLeaderReplicaCapacityException(
+ String.format(
+ "Not enough KV leader replica capacity. "
+ + "observedKvLeaderReplicaCount=%s, "
+ + "requestedKvLeaderReplicaCount=%s, "
+ + "kvLeaderReplicaCapacity=%s.",
+ observedCount, requestedKvLeaderReplicaCount,
capacity));
+ }
+ }
+ }
+
+ /** Replaces the observed KV leader replica count with a
CoordinatorContext-derived snapshot. */
+ public void updateObservedKvLeaderReplicaCount(long
observedKvLeaderReplicaCount) {
+ checkArgument(
+ observedKvLeaderReplicaCount >= 0,
+ "Observed KV leader replica count must be greater than or
equal to 0.");
+ this.observedKvLeaderReplicaCount = observedKvLeaderReplicaCount;
+ }
+
+ /** Returns the latest CoordinatorContext-derived KV leader replica count.
*/
+ public long getKvLeaderReplicaCount() {
+ return observedKvLeaderReplicaCount;
+ }
+
+ /** Returns the current cluster KV leader replica capacity, or -1 if
disabled. */
+ public long getKvLeaderReplicaCapacity() {
+ synchronized (lock) {
+ return getKvLeaderReplicaCapacityLocked();
+ }
+ }
+
+ @VisibleForTesting
+ long getKvLeaderReplicaMemoryReservedBytes() {
+ synchronized (lock) {
+ return kvLeaderReplicaMemoryReservedBytes;
+ }
+ }
+
+ private void registerMetrics(CoordinatorMetricGroup
coordinatorMetricGroup) {
+ coordinatorMetricGroup.gauge(
+ MetricNames.KV_LEADER_REPLICA_COUNT,
this::getKvLeaderReplicaCount);
+ coordinatorMetricGroup.gauge(
+ MetricNames.KV_LEADER_REPLICA_CAPACITY,
this::getKvLeaderReplicaCapacity);
+ }
+
+ private long getKvLeaderReplicaCapacityLocked() {
+ if (kvLeaderReplicaMemoryReservedBytes == 0) {
Review Comment:
Should copy to a local variable first, then access via the local variable.
Otherwise, there maybe in-consistency in this method.
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -319,6 +319,17 @@ public class ConfigOptions {
+ "This default is capped to reduce the
risk that an assignment znode exceeds "
+ "ZooKeeper's packet size limit.");
+ public static final ConfigOption<MemorySize>
KV_LEADER_REPLICA_MEMORY_RESERVED =
+ key("kv.leader-replica.memory-reserved")
+ .memoryType()
+ .defaultValue(MemorySize.parse("8mb"))
Review Comment:
I'm thinking that, should we disable it by default? I'm worried there are
some bad cases in different environment. And it can be a advanced feature that
can be enabled by users.
Btw, we should add documentation about this feature (when and how to
enable/disable/configure it). Can be a follow-up issue/PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
