You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@amoro.apache.org by ji...@apache.org on 2024/07/23 02:58:08 UTC
(amoro) branch master updated: [AMORO-3035] Support setting thrift client pool configs from url (#3036)
This is an automated email from the ASF dual-hosted git repository.
jinsongzhou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 855572b27 [AMORO-3035] Support setting thrift client pool configs from url (#3036)
855572b27 is described below
commit 855572b278bb783f344f50fd94c7b97bcc571eba
Author: ZhouJinsong <zh...@163.com>
AuthorDate: Tue Jul 23 10:58:05 2024 +0800
[AMORO-3035] Support setting thrift client pool configs from url (#3036)
* Parse client pool configs from url
* Add url parameter doc
* Set both maxMessageSize and maxFragmentSize for thrift client transport
---
.../org/apache/amoro/client/AmsClientPools.java | 18 +--
.../apache/amoro/client/OptimizingClientPools.java | 13 +--
.../java/org/apache/amoro/client/PoolConfig.java | 100 +++++++++++-----
.../org/apache/amoro/client/ThriftClientPool.java | 127 ++++++++-------------
.../org/apache/amoro/client/TestPoolConfig.java | 60 ++++++++++
docs/admin-guides/managing-optimizers.md | 31 ++++-
6 files changed, 211 insertions(+), 138 deletions(-)
diff --git a/amoro-core/src/main/java/org/apache/amoro/client/AmsClientPools.java b/amoro-core/src/main/java/org/apache/amoro/client/AmsClientPools.java
index e5fe8fff7..861d617a6 100644
--- a/amoro-core/src/main/java/org/apache/amoro/client/AmsClientPools.java
+++ b/amoro-core/src/main/java/org/apache/amoro/client/AmsClientPools.java
@@ -30,10 +30,6 @@ import org.apache.amoro.shade.thrift.org.apache.thrift.protocol.TProtocol;
/** Client pool cache for different ams server, sharing in jvm. */
public class AmsClientPools {
- private static final int CLIENT_POOL_MIN_IDLE = 0;
- private static final int CLIENT_POOL_MAX_IDLE = 5;
- private static final int CLIENT_POOL_MAX_WAIT_MS = 5000;
-
private static final LoadingCache<String, ThriftClientPool<AmoroTableMetastore.Client>>
CLIENT_POOLS = Caffeine.newBuilder().build(AmsClientPools::buildClientPool);
@@ -45,20 +41,16 @@ public class AmsClientPools {
CLIENT_POOLS.cleanUp();
}
+ @SuppressWarnings("unchecked")
private static ThriftClientPool<AmoroTableMetastore.Client> buildClientPool(String url) {
- PoolConfig poolConfig = new PoolConfig();
- poolConfig.setFailover(true);
- poolConfig.setMinIdle(CLIENT_POOL_MIN_IDLE);
- poolConfig.setMaxIdle(CLIENT_POOL_MAX_IDLE);
- poolConfig.setMaxWaitMillis(CLIENT_POOL_MAX_WAIT_MS);
+ PoolConfig<AmoroTableMetastore.Client> poolConfig =
+ (PoolConfig<AmoroTableMetastore.Client>) PoolConfig.forUrl(url);
return new ThriftClientPool<>(
url,
s -> {
TProtocol protocol = new TBinaryProtocol(s);
- AmoroTableMetastore.Client tableMetastore =
- new AmoroTableMetastore.Client(
- new TMultiplexedProtocol(protocol, Constants.THRIFT_TABLE_SERVICE_NAME));
- return tableMetastore;
+ return new AmoroTableMetastore.Client(
+ new TMultiplexedProtocol(protocol, Constants.THRIFT_TABLE_SERVICE_NAME));
},
c -> {
try {
diff --git a/amoro-core/src/main/java/org/apache/amoro/client/OptimizingClientPools.java b/amoro-core/src/main/java/org/apache/amoro/client/OptimizingClientPools.java
index 4cbf2df80..f03a78215 100644
--- a/amoro-core/src/main/java/org/apache/amoro/client/OptimizingClientPools.java
+++ b/amoro-core/src/main/java/org/apache/amoro/client/OptimizingClientPools.java
@@ -26,23 +26,22 @@ import org.apache.amoro.shade.thrift.org.apache.thrift.TException;
import org.apache.amoro.shade.thrift.org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.amoro.shade.thrift.org.apache.thrift.protocol.TMultiplexedProtocol;
+import java.util.Objects;
+
/** Client pool cache for different ams optimize server, sharing in jvm. */
public class OptimizingClientPools {
- private static final int CLIENT_POOL_MIN_IDLE = 1;
- private static final int CLIENT_POOL_MAX_IDLE = 5;
private static final LoadingCache<String, ThriftClientPool<OptimizingService.Client>>
CLIENT_POOLS = Caffeine.newBuilder().build(OptimizingClientPools::buildClient);
public static OptimizingService.Iface getClient(String metastoreUrl) {
- return CLIENT_POOLS.get(metastoreUrl).iface();
+ return Objects.requireNonNull(CLIENT_POOLS.get(metastoreUrl)).iface();
}
+ @SuppressWarnings("unchecked")
private static ThriftClientPool<OptimizingService.Client> buildClient(String url) {
- PoolConfig poolConfig = new PoolConfig();
- poolConfig.setFailover(true);
- poolConfig.setMinIdle(CLIENT_POOL_MIN_IDLE);
- poolConfig.setMaxIdle(CLIENT_POOL_MAX_IDLE);
+ PoolConfig<OptimizingService.Client> poolConfig =
+ (PoolConfig<OptimizingService.Client>) PoolConfig.forUrl(url);
poolConfig.setMaxTotal(-1);
return new ThriftClientPool<>(
url,
diff --git a/amoro-core/src/main/java/org/apache/amoro/client/PoolConfig.java b/amoro-core/src/main/java/org/apache/amoro/client/PoolConfig.java
index 259bdfcec..cade1831c 100644
--- a/amoro-core/src/main/java/org/apache/amoro/client/PoolConfig.java
+++ b/amoro-core/src/main/java/org/apache/amoro/client/PoolConfig.java
@@ -18,47 +18,85 @@
package org.apache.amoro.client;
+import org.apache.amoro.shade.thrift.org.apache.thrift.TServiceClient;
+import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import org.apache.http.client.utils.URLEncodedUtils;
-public class PoolConfig extends GenericObjectPoolConfig {
+import java.lang.reflect.InvocationTargetException;
+import java.net.URI;
+import java.nio.charset.Charset;
+import java.time.Duration;
- private int timeout = 0;
+public class PoolConfig<T extends TServiceClient> extends GenericObjectPoolConfig<ThriftClient<T>> {
- private boolean failover = false;
+ private static final int MIN_IDLE_DEFAULT = 0;
+ private static final int MAX_IDLE_DEFAULT = 5;
+ private static final int MAX_WAIT_MS_DEFAULT = 5000;
- /**
- * get default connection socket timeout (default 0, means not timeout)
- *
- * @return
- */
- public int getTimeout() {
- return timeout;
+ private int connectTimeout = 0;
+ private int socketTimeout = 0;
+ private int maxMessageSize = 100 * 1024 * 1024; // 100MB
+ private boolean autoReconnect = true;
+ private int maxReconnects = 5;
+
+ public PoolConfig() {
+ setMinIdle(MIN_IDLE_DEFAULT);
+ setMaxIdle(MAX_IDLE_DEFAULT);
+ setMaxWait(Duration.ofMillis(MAX_WAIT_MS_DEFAULT));
+ }
+
+ public int getConnectTimeout() {
+ return connectTimeout;
+ }
+
+ public void setConnectTimeout(int connectTimeout) {
+ this.connectTimeout = connectTimeout;
+ }
+
+ public int getSocketTimeout() {
+ return socketTimeout;
+ }
+
+ public void setSocketTimeout(int socketTimeout) {
+ this.socketTimeout = socketTimeout;
+ }
+
+ public int getMaxMessageSize() {
+ return maxMessageSize;
+ }
+
+ public void setMaxMessageSize(int maxMessageSize) {
+ this.maxMessageSize = maxMessageSize;
+ }
+
+ public boolean isAutoReconnect() {
+ return autoReconnect;
+ }
+
+ public void setAutoReconnect(boolean autoReconnect) {
+ this.autoReconnect = autoReconnect;
}
- /**
- * set default connection socket timeout
- *
- * @param timeout timeout millis
- */
- public void setTimeout(int timeout) {
- this.timeout = timeout;
+ public int getMaxReconnects() {
+ return maxReconnects;
}
- /**
- * get connect to next service if one service fail(default false)
- *
- * @return
- */
- public boolean isFailover() {
- return failover;
+ public void setMaxReconnects(int maxReconnects) {
+ this.maxReconnects = maxReconnects;
}
- /**
- * set connect to next service if one service fail
- *
- * @param failover
- */
- public void setFailover(boolean failover) {
- this.failover = failover;
+ public static PoolConfig<?> forUrl(String url) {
+ PoolConfig<?> poolConfig = new PoolConfig<>();
+ URLEncodedUtils.parse(URI.create(url), Charset.defaultCharset())
+ .forEach(
+ pair -> {
+ try {
+ BeanUtils.setProperty(poolConfig, pair.getName(), pair.getValue());
+ } catch (IllegalAccessException | InvocationTargetException e) {
+ throw new RuntimeException("Parse url parameters failed", e);
+ }
+ });
+ return poolConfig;
}
}
diff --git a/amoro-core/src/main/java/org/apache/amoro/client/ThriftClientPool.java b/amoro-core/src/main/java/org/apache/amoro/client/ThriftClientPool.java
index e8f6ea6d7..bfdfe4439 100644
--- a/amoro-core/src/main/java/org/apache/amoro/client/ThriftClientPool.java
+++ b/amoro-core/src/main/java/org/apache/amoro/client/ThriftClientPool.java
@@ -18,6 +18,7 @@
package org.apache.amoro.client;
+import org.apache.amoro.shade.thrift.org.apache.thrift.TConfiguration;
import org.apache.amoro.shade.thrift.org.apache.thrift.transport.TSocket;
import org.apache.amoro.shade.thrift.org.apache.thrift.transport.TTransport;
import org.apache.amoro.shade.thrift.org.apache.thrift.transport.TTransportException;
@@ -31,39 +32,25 @@ import org.slf4j.LoggerFactory;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Proxy;
-import java.util.List;
-import java.util.Random;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Collectors;
public class ThriftClientPool<
T extends org.apache.amoro.shade.thrift.org.apache.thrift.TServiceClient> {
-
private static final Logger LOG = LoggerFactory.getLogger(ThriftClientPool.class);
- // for thrift connects
- private static final int retries = 5;
- private static final int retryInterval = 2000;
- private static final int maxMessageSize = 100 * 1024 * 1024;
+ private static final int RECONNECT_INTERVAL = 2000;
+ private static final int BORROW_ATTEMPTS = 5;
+
private final ThriftClientFactory clientFactory;
private final ThriftPingFactory pingFactory;
private final GenericObjectPool<ThriftClient<T>> pool;
- private final PoolConfig poolConfig;
- private final String serviceName;
- private final String url;
- private final boolean serviceReset = false;
+ private final PoolConfig<T> poolConfig;
- /**
- * Construct a new pool using
- *
- * @param url
- * @param factory
- * @param config
- */
public ThriftClientPool(
String url,
ThriftClientFactory factory,
ThriftPingFactory pingFactory,
- PoolConfig config,
+ PoolConfig<T> config,
String serviceName) {
if (url == null || url.isEmpty()) {
throw new IllegalArgumentException("url is empty!");
@@ -75,14 +62,12 @@ public class ThriftClientPool<
throw new IllegalArgumentException("config is empty!");
}
- this.url = url;
this.clientFactory = factory;
this.pingFactory = pingFactory;
this.poolConfig = config;
// test if config change
this.poolConfig.setTestOnReturn(true);
this.poolConfig.setTestOnBorrow(true);
- this.serviceName = serviceName;
this.pool =
new GenericObjectPool<>(
new BasePooledObjectFactory<ThriftClient<T>>() {
@@ -99,48 +84,53 @@ public class ThriftClientPool<
transport.open();
} catch (TTransportException e) {
LOG.warn(
- "transport open fail service: host={}, port={}",
+ "Transport open failed, service address: {}:{}",
serviceInfo.getHost(),
serviceInfo.getPort());
- if (poolConfig.isFailover()) {
- for (int i = 0; i < 5; i++) {
+ if (poolConfig.isAutoReconnect() && poolConfig.getMaxReconnects() > 0) {
+ for (int i = 0; i < poolConfig.getMaxReconnects(); i++) {
try {
amsThriftUrl = AmsThriftUrl.parse(url, serviceName);
serviceInfo.setHost(amsThriftUrl.host());
serviceInfo.setPort(amsThriftUrl.port());
transport = getTransport(serviceInfo); // while break here
LOG.info(
- "failover to next service host={}, port={}",
+ "Reconnecting service address: {}:{}",
serviceInfo.getHost(),
serviceInfo.getPort());
transport.open();
break;
} catch (TTransportException e2) {
LOG.warn(
- "transport open fail service: host={}, port={}",
+ "Reconnected service address: {}:{} failed",
serviceInfo.getHost(),
- serviceInfo.getPort());
+ serviceInfo.getPort(),
+ e2);
}
- Thread.sleep(retryInterval);
+ TimeUnit.MILLISECONDS.sleep(RECONNECT_INTERVAL);
}
if (!transport.isOpen()) {
throw new ConnectionFailException(
- "connect error after try 5 times, last connect is: host="
- + serviceInfo.getHost()
- + ", ip="
- + serviceInfo.getPort(),
+ String.format(
+ "Connected error after %d retries, last address: %s:%d",
+ poolConfig.getMaxReconnects(),
+ serviceInfo.getHost(),
+ serviceInfo.getPort()),
e);
}
} else {
throw new ConnectionFailException(
- "host=" + serviceInfo.getHost() + ", ip=" + serviceInfo.getPort(), e);
+ String.format(
+ "Connected error, address: %s:%d",
+ serviceInfo.getHost(), serviceInfo.getPort()),
+ e);
}
}
ThriftClient<T> client =
new ThriftClient<>(clientFactory.createClient(transport), pool, serviceInfo);
- LOG.debug("create new object for pool {}", client);
+ LOG.debug("created new thrift pool for url:{}", url);
return client;
}
@@ -163,46 +153,20 @@ public class ThriftClientPool<
if (serviceInfo == null) {
throw new NoBackendServiceException();
}
-
- TTransport transport = null;
- if (poolConfig.getTimeout() > 0) {
- transport =
- new TFramedTransport(
- new TSocket(serviceInfo.getHost(), serviceInfo.getPort(), poolConfig.getTimeout()),
- maxMessageSize);
- } else {
- transport =
- new TFramedTransport(
- new TSocket(serviceInfo.getHost(), serviceInfo.getPort()), maxMessageSize);
- }
- return transport;
- }
-
- /**
- * get a random service
- *
- * @param serviceList
- * @return
- */
- private ServiceInfo getRandomService(List<ServiceInfo> serviceList) {
- if (serviceList == null || serviceList.size() == 0) {
- return null;
- }
- return serviceList.get(new Random().nextInt(serviceList.size()));
- }
-
- private List<ServiceInfo> removeFailService(List<ServiceInfo> list, ServiceInfo serviceInfo) {
- LOG.info(
- "remove service from current service list: host={}, port={}",
- serviceInfo.getHost(),
- serviceInfo.getPort());
- return list.stream() //
- .filter(si -> !serviceInfo.equals(si)) //
- .collect(Collectors.toList());
+ TConfiguration configuration = new TConfiguration();
+ configuration.setMaxMessageSize(poolConfig.getMaxMessageSize());
+ return new TFramedTransport(
+ new TSocket(
+ new TConfiguration(),
+ serviceInfo.getHost(),
+ serviceInfo.getPort(),
+ poolConfig.getConnectTimeout(),
+ poolConfig.getSocketTimeout()),
+ poolConfig.getMaxMessageSize());
}
/**
- * get a client's IFace from pool
+ * Get a client's IFace from pool
*
* <ul>
* <li><span style="color:red">Important: Iface is totally generated by thrift, a
@@ -210,11 +174,10 @@ public class ThriftClientPool<
* <li><span style="color:red">Limitation: The return object can only used once.</span>
* </ul>
*
- * @return
- * @throws ThriftException
- * @throws NoBackendServiceException if {@link PoolConfig#setFailover(boolean)} is set and no
+ * @return thrift service interface
+ * @throws NoBackendServiceException if {@link PoolConfig#setAutoReconnect(boolean)} is set and no
* service can connect to
- * @throws ConnectionFailException if {@link PoolConfig#setFailover(boolean)} not set and
+ * @throws ConnectionFailException if {@link PoolConfig#setAutoReconnect(boolean)} not set and
* connection fail
* @throws IllegalStateException if call method on return object twice
*/
@@ -222,14 +185,14 @@ public class ThriftClientPool<
public <X> X iface() {
ThriftClient<T> client = null;
int attempt;
- for (attempt = 0; attempt < retries; ++attempt) {
+ for (attempt = 0; attempt < BORROW_ATTEMPTS; ++attempt) {
try {
client = pool.borrowObject();
if (client.isDisConnected() || !pingFactory.ping(client.iface())) {
if (attempt > 1) {
// if attempt > 1, it means the server is maybe restarting, so we should wait a while
- LOG.warn("maybe server is restarting, wait a while");
- Thread.sleep(retryInterval);
+ LOG.warn("Maybe server is restarting, wait a while");
+ Thread.sleep(RECONNECT_INTERVAL);
}
pool.invalidateObject(client);
pool.clear();
@@ -240,10 +203,10 @@ public class ThriftClientPool<
if (e instanceof ThriftException) {
throw (ThriftException) e;
}
- throw new ThriftException("Get client from pool failed.", e);
+ throw new ThriftException("Get client from pool failed", e);
}
}
- if (attempt >= retries) {
+ if (attempt >= BORROW_ATTEMPTS) {
throw new ThriftException("Client can not connect.");
}
AtomicBoolean returnToPool = new AtomicBoolean(false);
@@ -254,7 +217,7 @@ public class ThriftClientPool<
finalClient.iface().getClass().getInterfaces(),
(proxy, method, args) -> {
if (returnToPool.get()) {
- throw new IllegalStateException("Object returned via iface can only used once!");
+ throw new IllegalStateException("Client borrowed from pool can only used once!");
}
boolean success = false;
try {
diff --git a/amoro-core/src/test/java/org/apache/amoro/client/TestPoolConfig.java b/amoro-core/src/test/java/org/apache/amoro/client/TestPoolConfig.java
new file mode 100644
index 000000000..a36f9ea79
--- /dev/null
+++ b/amoro-core/src/test/java/org/apache/amoro/client/TestPoolConfig.java
@@ -0,0 +1,60 @@
+/*
+ * 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.amoro.client;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestPoolConfig {
+
+ @Test
+ public void testConstructFromUrl() {
+ String url =
+ "thrift://127.0.0.1:1261?connectTimeout=5000&socketTimeout=6000&maxMessageSize=100&autoReconnect=false"
+ + "&maxReconnects=3&minIdle=10&maxIdle=10&maxWaitMillis=500";
+ PoolConfig<?> poolConfig = PoolConfig.forUrl(url);
+
+ Assert.assertEquals(5000, poolConfig.getConnectTimeout());
+ Assert.assertEquals(6000, poolConfig.getSocketTimeout());
+ Assert.assertEquals(100, poolConfig.getMaxMessageSize());
+ Assert.assertFalse(poolConfig.isAutoReconnect());
+ Assert.assertEquals(3, poolConfig.getMaxReconnects());
+ Assert.assertEquals(10, poolConfig.getMinIdle());
+ Assert.assertEquals(10, poolConfig.getMaxIdle());
+ Assert.assertEquals(500, poolConfig.getMaxWaitMillis());
+ }
+
+ @Test
+ public void tetUrlParameterNameError() {
+ // We will ignore parameters with unknown name
+ String url = "thrift://127.0.0.1:1261?connectTimeouts=300";
+ PoolConfig<?> poolConfig = PoolConfig.forUrl(url);
+ Assert.assertEquals(0, poolConfig.getConnectTimeout());
+ }
+
+ @Test
+ public void testUrlFormatError() {
+ String url = "thrift://127.0.0.1:1261?connectTimeout=5000& ";
+ Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> {
+ PoolConfig.forUrl(url);
+ });
+ }
+}
diff --git a/docs/admin-guides/managing-optimizers.md b/docs/admin-guides/managing-optimizers.md
index 2da26a903..afd6caed1 100644
--- a/docs/admin-guides/managing-optimizers.md
+++ b/docs/admin-guides/managing-optimizers.md
@@ -33,11 +33,18 @@ The optimizer is the execution unit for performing self-optimizing tasks on a ta
* Optimizer: The specific unit that performs optimizing tasks, usually with multiple concurrent units.
## Optimizer container
-Before using self-optimizing, you need to configure the container information in the configuration file. Optimizer container represents a specific set of runtime environment configuration, and the scheduling scheme of optimizer in that runtime environment. The container includes three types: flink, local, and external.
+Before start exploring self-optimizing, you need to configure the container information in the configuration file. Optimizer container represents a specific set of runtime environment configuration. The supported container types include: local, kubernetes, flink, spark, and external.
### Local container
Local container is a way to start Optimizer by local process and supports multi-threaded execution of Optimizer tasks. It is recommended to be used only in demo or local deployment scenarios. If the environment variable for jdk is not configured, the user can configure java_home to point to the jdk root directory. If already configured, this configuration item can be ignored.
+Local container support the following properties:
+
+| Property Name | Required | Default Value | Description |
+|-----------------------|----------|---------------|---------------------------------------------------------------------------------------------------------------|
+| ams-optimizing-uri | false | N/A | URI of AMS thrift self-optimizing endpoint. This could be used if the ams.server-expose-host is not available |
+| export.JAVA_HOME | false | N/A | Java runtime location |
+
```yaml
containers:
- name: localContainer
@@ -46,13 +53,27 @@ containers:
export.JAVA_HOME: "/opt/java" # JDK environment
```
+The format for optimizing URI is `thrift://{host}:{port}?parameter1=value2¶meter2=value2`.
+The supported parameters include:
+
+| Parameter Name | efault Value | Description |
+|----------------|-------------------|------------------------------------------------------------|
+| autoReconnect | true | If reconnect the server when the connection is broken |
+| maxReconnects | 5 | Retry times when reconnecting |
+| connectTimeout | 0 (Forever) | Timeout in milliseconds when connecting the server |
+| socketTimeout | 0 (Forever) | Timeout in milliseconds when communicating with the server |
+| maxMessageSize | 104856600 (100MB) | Max message size when communicating with the server |
+| maxMessageSize | 104856600 (100MB) | Max message size when communicating with the server |
+| minIdle | 0 | Minimal idle clients in the pool |
+| maxIdle | 5 | Maximal idle clients in the pool |
+
### Kubernetes container
-Kubernetes Optimizer Container is a way to start Optimizer On K8s with standalone Optimizer.
+Kubernetes container is a way to start Optimizer On K8s with standalone Optimizer.
To use flink container, you need to add a new container configuration.
with container-impl as `org.apache.amoro.server.manager.KubernetesOptimizerContainer`
-KubernetesOptimizerContainer support the following properties:
+Kubernetes container support the following properties:
| Property Name | Required | Default Value | Description |
@@ -81,7 +102,7 @@ Flink container is a way to start Optimizer through Flink jobs. With Flink, you
on yarn clusters or kubernetes clusters to support large-scale data scenarios. To use flink container,
you need to add a new container configuration. with container-impl as `org.apache.amoro.server.manager.FlinkOptimizerContainer`
-FlinkOptimizerContainer support the following properties:
+Flink container support the following properties:
| Property Name | Required | Default Value | Description |
|---------------------------|----------|------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
@@ -161,7 +182,7 @@ Spark container is another way to start Optimizer through Spark jobs. With Spark
on yarn clusters or kubernetes clusters to support large-scale data scenarios. To use spark container,
you need to add a new container configuration. with container-impl as `org.apache.amoro.server.manager.SparkOptimizerContainer`
-SparkOptimizerContainer support the following properties:
+Spark container support the following properties:
| Property Name | Required | Default Value | Description |
|-------------------------|----------|---------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|