jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

!1463 Introducing Failure Retry Policies 

Merge pull request !1463 from ahanapradhan/newpr
This commit is contained in:
i-robot 2022-05-17 10:36:48 +00:00 committed by Gitee
parent e21b036efb a36635f676
commit 56305dcfc2
No known key found for this signature in database
GPG Key ID: 173E9B9CA92EEF8F
33 changed files with 1216 additions and 386 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
hetu-docs/en/admin/properties.md
View File

@ -368,29 +368,6 @@ Exchanges transfer data between openLooKeng nodes for different stages of a quer
>
> Increasing the value may improve network throughput if there is high latency. Decreasing the value may improve query performance for large clusters as it reduces skew due to the exchange client buffer holding responses for more tasks (rather than hold more data from fewer tasks).
### `exchange.max-error-duration`
> - **Type:** `duration`
> - **Minimum value:** `1m`
> - **Default value:** `7m`
>
> The maximum amount of time coordinator waits for inter-task related errors to be resolved before it's considered a failure.
### `exchange.is-timeout-failure-detection-enabled`
> - **Type:** `boolean`
> - **Default value:** `true`
>
> The failure detection mechanism in use. Default is timeout based failure detection. Otherwise, i.e. when this property is set to false, maximum retry based failure detection mechanism is enabled.
>
### `exchange.max-retry-count`
> - **Type:** `integer`
> - **Default value:** `100`
>
> The maximum number of retry for failed task performed by the coordinator before consulting the failure detector module about the remote node status. If the remote node status is failed as per the failure detector module, it is considered as a permanent failure. This parameter is the minimum count which is required to decide, not necessarily the exact count. Based on the cluster size, load on the cluster the exact count may vary slightly. This property is used only when exchange.is-timeout-failure-detection-enabled is set to false. This value needs to be at least 100 to take effect.
### `sink.max-buffer-size`
> - **Type:** `data size`
@ -398,6 +375,52 @@ Exchanges transfer data between openLooKeng nodes for different stages of a quer
>
> Output buffer size for task data that is waiting to be pulled by upstream tasks. If the task output is hash partitioned, then the buffer will be shared across all of the partitioned consumers. Increasing this value may improve network throughput for data transferred between stages if the network has high latency or if there are many nodes in the cluster.
## Failure Recovery handling Properties
### `failure.recovery.retry.profile`
> - **Type:** `String`
> - **Default value:** `default`
>
> This property defines the failure detection profile used to determine if failure has happened for a http client. The value `<profile-name>` set for this property has to correspond to `<profile-name>.properties` file in `etc/failure-retry-policy/`. In case no such profile is available, and this property is not set, "default" profile is used.
> For example, `failure.recovery.retry.profile="test"` requires `test.properties` file to be present in `etc/failure-retry-policy`.
> The file `test.properties` must contain `failure.recovery.retry.type` specified.
### `failure.recovery.retry.type`
> - **Type:** `String`
> - **Default value:** `timeout`
>
> The failure detection mechanism in use. Default is timeout based failure detection.
>
#### `timeout` based failure detection.
> Using this mechanism, HTTP client failures are retried for a specific duration before considering it as a permanent failure.
> Additional properties `max.error.duration` can be defined for this type of failure detection.
>
#### `max-retry` based failure detection.
> Using this mechanism, HTTP client failures are retried for a specific number of times before considering it as a permanent failure.
> Additional properties `max.retry.count` and `max.error.duration` can be defined for this type of failure detection.
> Using this type of failure detection is configured to be used, `max.retry.count` times retry is performed before consulting the failure detector module. When the remote node is failed as per the failure detector module, HTTP client considers it a permanent failure. Otherwise, i.e. When remote worker node is alive but not sending response, retry happens for `max.error.duration` before considering it as permanent failure.
### `max.error.duration`
> - **Type:** `duration`
> - **Default value:** `300s`
>
> The maximum amount of time coordinator waits for inter-task related errors to be resolved before it's considered a permanent failure.
### `max.retry.count`
> - **Type:** `integer`
> - **Default value:** `100`
>
> The maximum number of retry for failed task performed by the coordinator before consulting the failure detector module about the remote node status.
> This parameter is the minimum count before consulting the failure detection module. Hence, the actual number of failures may vary slightly based on the cluster size, and load on the cluster.
> This property is used only for `max-retry` based failure detection profiles.
> The minimum value for this parameter is 100.
## Task Properties
### `task.concurrency`

37
presto-main/src/main/java/io/prestosql/failuredetector/AbstractFailureRetryPolicy.java Normal file
View File

@ -0,0 +1,37 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.failuredetector;
import io.prestosql.spi.HostAddress;
import io.prestosql.spi.failuredetector.FailureRetryPolicy;
import io.prestosql.spi.failuredetector.IBackoff;
public abstract class AbstractFailureRetryPolicy
implements FailureRetryPolicy
{
private final IBackoff backoff;
public AbstractFailureRetryPolicy(IBackoff backoff)
{
this.backoff = backoff;
}
public IBackoff getBackoff()
{
return this.backoff;
}
public abstract boolean hasFailed(HostAddress address);
}

203
presto-main/src/main/java/io/prestosql/failuredetector/FailureDetectorManager.java Normal file
View File

@ -0,0 +1,203 @@
/*
* Licensed 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 io.prestosql.failuredetector;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Ticker;
import io.airlift.log.Logger;
import io.prestosql.spi.classloader.ThreadContextClassLoader;
import io.prestosql.spi.failuredetector.FailureRetryFactory;
import io.prestosql.spi.failuredetector.FailureRetryPolicy;
import javax.inject.Inject;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import static com.google.common.base.Preconditions.checkState;
import static java.util.Objects.requireNonNull;
public class FailureDetectorManager
{
private static final Logger LOG = Logger.get(FailureDetectorManager.class);
public static final String FD_RETRY_CONFIG_DIR = "etc/failure-retry-policy/";
private final String failureRetryPolicyConfig;
private static final List<FailureDetector> failureDetectors = new ArrayList<>();
private static final Map<String, FailureRetryFactory> failureRetryFactories = new ConcurrentHashMap<>();
private static final Map<String, Properties> availableFrConfigs = new ConcurrentHashMap<>();
@VisibleForTesting
public FailureDetectorManager(FailureDetector failureDetector, String maxErrorDuration)
{
requireNonNull(failureDetector, "failureDetector is null");
failureDetectors.add(failureDetector);
Properties defaultProfile = new Properties();
defaultProfile.setProperty(FailureRetryPolicy.FD_RETRY_TYPE, FailureRetryPolicy.TIMEOUT);
defaultProfile.setProperty(FailureRetryPolicy.MAX_TIMEOUT_DURATION, maxErrorDuration);
availableFrConfigs.put(FailureRetryConfig.DEFAULT_CONFIG_NAME, defaultProfile);
this.failureRetryPolicyConfig = FailureRetryConfig.DEFAULT_CONFIG_NAME;
}
@Inject
public FailureDetectorManager(FailureRetryConfig cfg, FailureDetector failureDetector)
{
requireNonNull(failureDetector, "failureDetector is null");
requireNonNull(cfg, "config is null");
failureDetectors.add(failureDetector);
Properties defaultProfile = new Properties();
defaultProfile.setProperty(FailureRetryPolicy.FD_RETRY_TYPE, FailureRetryPolicy.TIMEOUT);
defaultProfile.setProperty(FailureRetryPolicy.MAX_TIMEOUT_DURATION, FailureRetryPolicy.DEFAULT_TIMEOUT_DURATION);
availableFrConfigs.putIfAbsent(cfg.getFailureRetryPolicyProfile(), defaultProfile);
this.failureRetryPolicyConfig = cfg.getFailureRetryPolicyProfile();
}
@VisibleForTesting
public static synchronized Map<String, Properties> getAvailableFrConfigs()
{
return availableFrConfigs;
}
public static synchronized FailureDetector getDefaultFailureDetector()
{
return failureDetectors.get(0);
}
public static synchronized void addFailureRetryFactory(FailureRetryFactory factory)
{
failureRetryFactories.putIfAbsent(factory.getName(), factory);
}
@VisibleForTesting
public static synchronized Map<String, FailureRetryFactory> getFailureRetryFactories()
{
return failureRetryFactories;
}
@VisibleForTesting
public static synchronized void removeallFrConfigs()
{
availableFrConfigs.clear();
}
@VisibleForTesting
public static synchronized void addFrConfigs(String profileName, Properties properties)
{
availableFrConfigs.putIfAbsent(profileName, properties);
}
public String getFailureRetryPolicyUserProfile()
{
return this.failureRetryPolicyConfig;
}
public void loadFactoryConfigs()
throws IOException
{
LOG.info("-- Available failure retry policy factories: %s --", failureRetryFactories.keySet().toString());
File configDir = new File(FD_RETRY_CONFIG_DIR);
if (!configDir.exists() || !configDir.isDirectory()) {
LOG.info("-- failure retry policy configs not found. Skipped loading --");
return;
}
String[] failureRetries = configDir.list();
if (failureRetries == null || failureRetries.length == 0) {
LOG.info("-- no retry policy set. Default failure retry policy will be used. --");
return;
}
for (String fileName : failureRetries) {
if (fileName.endsWith(".properties")) {
String configName = fileName.replaceAll("\\.properties", "");
File configFile = new File(FD_RETRY_CONFIG_DIR + fileName);
Properties properties = loadProperties(configFile);
String configType = properties.getProperty(FailureRetryPolicy.FD_RETRY_TYPE);
checkState(configType != null, "%s must be specified in %s",
FailureRetryPolicy.FD_RETRY_TYPE, configFile.getCanonicalPath());
checkState(failureRetryFactories.containsKey(configType),
"Factory for failure retry policy type %s not found", configType);
availableFrConfigs.put(configName, properties);
LOG.info(String.format("Loaded '%s' failure retry policy config '%s'", configType, configName));
LOG.info(String.format("-- Loaded failure retry profiles: %s --",
availableFrConfigs.keySet().toString()));
}
}
}
private Properties loadProperties(File configFile)
throws IOException
{
Properties properties = new Properties();
try (InputStream in = new FileInputStream(configFile)) {
properties.load(in);
}
return properties;
}
public FailureRetryPolicy getFailureRetryPolicy(String name)
{
Properties frConfig = availableFrConfigs.get(name);
LOG.debug("Profile name: " + name + ", retry type: " + frConfig.getProperty(FailureRetryPolicy.FD_RETRY_TYPE));
return getFailureRetryPolicy(frConfig);
}
private String checkProperty(Properties properties, String key)
{
String val = properties.getProperty(key);
LOG.debug("Key: " + key + ", value: " + val);
if (val == null) {
throw new IllegalArgumentException(String.format("Configuration entry '%s' must be specified", key));
}
return val;
}
private FailureRetryPolicy getFailureRetryPolicy(Properties properties)
{
String type = checkProperty(properties, FailureRetryPolicy.FD_RETRY_TYPE);
checkState(failureRetryFactories.containsKey(type),
"Factory for failure retry policy type %s not found", type);
FailureRetryFactory factory = failureRetryFactories.get(type);
try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(factory.getClass().getClassLoader())) {
return factory.getFailureRetryPolicy(properties);
}
}
@VisibleForTesting
public FailureRetryPolicy getFailureRetryPolicy(String name, Ticker ticker)
{
Properties frConfig = availableFrConfigs.get(name);
String type = checkProperty(frConfig, FailureRetryPolicy.FD_RETRY_TYPE);
checkState(failureRetryFactories.containsKey(type),
"Factory for failure retry policy type %s not found", type);
FailureRetryFactory factory = failureRetryFactories.get(type);
try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(factory.getClass().getClassLoader())) {
return factory.getFailureRetryPolicy(frConfig, ticker);
}
}
}

29
presto-main/src/main/java/io/prestosql/failuredetector/FailureDetectorPlugin.java Normal file
View File

@ -0,0 +1,29 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.failuredetector;
import com.google.common.collect.ImmutableList;
import io.prestosql.spi.Plugin;
import io.prestosql.spi.failuredetector.FailureRetryFactory;
public class FailureDetectorPlugin
implements Plugin
{
@Override
public Iterable<FailureRetryFactory> getFailureRetryFactory()
{
return ImmutableList.of(new TimeoutFailureRetryFactory(), new MaxRetryFailureRetryFactory());
}
}

39
presto-main/src/main/java/io/prestosql/failuredetector/FailureRetryConfig.java Normal file
View File

@ -0,0 +1,39 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.failuredetector;
import io.airlift.configuration.Config;
import io.prestosql.spi.failuredetector.FailureRetryPolicy;
import javax.validation.constraints.NotNull;
public class FailureRetryConfig
{
public static final String DEFAULT_CONFIG_NAME = "default";
private String failureRetryPolicyConfig = DEFAULT_CONFIG_NAME;
@Config(FailureRetryPolicy.FD_RETRY_PROFILE)
public FailureRetryConfig setFailureRetryPolicyProfile(String profileName)
{
this.failureRetryPolicyConfig = profileName;
return this;
}
@NotNull
public String getFailureRetryPolicyProfile()
{
return failureRetryPolicyConfig;
}
}

28
presto-main/src/main/java/io/prestosql/failuredetector/FailureRetryPolicyFactory.java Normal file
View File

@ -0,0 +1,28 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.failuredetector;
import com.google.common.base.Ticker;
import io.prestosql.spi.failuredetector.FailureRetryFactory;
import java.util.Properties;
public abstract class FailureRetryPolicyFactory
implements FailureRetryFactory
{
public abstract AbstractFailureRetryPolicy getFailureRetryPolicy(Properties properties);
public abstract AbstractFailureRetryPolicy getFailureRetryPolicy(Properties properties, Ticker ticker);
}

41
presto-main/src/main/java/io/prestosql/failuredetector/MaxRetryFailureRetryConfig.java Normal file
View File

@ -0,0 +1,41 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.failuredetector;
import io.prestosql.spi.failuredetector.FailureRetryPolicy;
import javax.validation.constraints.Min;
import java.util.Properties;
public class MaxRetryFailureRetryConfig
{
private String maxRetryCount;
public MaxRetryFailureRetryConfig(Properties properties)
{
this.maxRetryCount = properties.getProperty(FailureRetryPolicy.MAX_RETRY_COUNT);
}
@Min(100)
public int getMaxRetryCount()
{
if (maxRetryCount == null) {
return FailureRetryPolicy.DEFAULT_RETRY_COUNT;
}
int count = Integer.parseInt(maxRetryCount);
return count;
}
}

42
presto-main/src/main/java/io/prestosql/failuredetector/MaxRetryFailureRetryFactory.java Normal file
View File

@ -0,0 +1,42 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.failuredetector;
import com.google.common.base.Ticker;
import io.prestosql.spi.failuredetector.FailureRetryPolicy;
import java.util.Properties;
public class MaxRetryFailureRetryFactory
extends FailureRetryPolicyFactory
{
@Override
public AbstractFailureRetryPolicy getFailureRetryPolicy(Properties properties)
{
return new MaxRetryFailureRetryPolicy(new MaxRetryFailureRetryConfig(properties));
}
@Override
public AbstractFailureRetryPolicy getFailureRetryPolicy(Properties properties, Ticker ticker)
{
return new MaxRetryFailureRetryPolicy(new MaxRetryFailureRetryConfig(properties), ticker);
}
@Override
public String getName()
{
return FailureRetryPolicy.MAXRETRY;
}
}

52
presto-main/src/main/java/io/prestosql/failuredetector/MaxRetryFailureRetryPolicy.java Normal file
View File

@ -0,0 +1,52 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.failuredetector;
import com.google.common.base.Ticker;
import io.airlift.log.Logger;
import io.prestosql.server.remotetask.MaxRetryBackoff;
import io.prestosql.spi.HostAddress;
public class MaxRetryFailureRetryPolicy
extends AbstractFailureRetryPolicy
{
private static final Logger log = Logger.get(MaxRetryFailureRetryPolicy.class);
private MaxRetryFailureRetryConfig config;
FailureDetector failureDetector;
public MaxRetryFailureRetryPolicy(MaxRetryFailureRetryConfig config)
{
super(new MaxRetryBackoff(config.getMaxRetryCount()));
this.failureDetector = FailureDetectorManager.getDefaultFailureDetector();
this.config = config;
}
public MaxRetryFailureRetryPolicy(MaxRetryFailureRetryConfig config, Ticker ticker)
{
super(new MaxRetryBackoff(config.getMaxRetryCount(), ticker));
this.failureDetector = FailureDetectorManager.getDefaultFailureDetector();
this.config = config;
}
@Override
public boolean hasFailed(HostAddress address)
{
FailureDetector.State remoteHostState = failureDetector.getState(address);
log.debug("remote node state is: " + remoteHostState.toString());
return (((MaxRetryBackoff) getBackoff()).maxRetryDone() && !FailureDetector.State.ALIVE.equals(remoteHostState))
|| ((MaxRetryBackoff) getBackoff()).timeout();
}
}

38
presto-main/src/main/java/io/prestosql/failuredetector/TimeoutFailureRetryConfig.java Normal file
View File

@ -0,0 +1,38 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.failuredetector;
import io.airlift.units.Duration;
import io.prestosql.spi.failuredetector.FailureRetryPolicy;
import java.util.Properties;
public class TimeoutFailureRetryConfig
{
private final String maxTimeoutDuration;
public TimeoutFailureRetryConfig(Properties properties)
{
this.maxTimeoutDuration = properties.getProperty(FailureRetryPolicy.MAX_TIMEOUT_DURATION);
}
public Duration getMaxTimeoutDuration()
{
if (maxTimeoutDuration == null) {
return Duration.valueOf(FailureRetryPolicy.DEFAULT_TIMEOUT_DURATION);
}
return Duration.valueOf(maxTimeoutDuration);
}
}

42
presto-main/src/main/java/io/prestosql/failuredetector/TimeoutFailureRetryFactory.java Normal file
View File

@ -0,0 +1,42 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.failuredetector;
import com.google.common.base.Ticker;
import io.prestosql.spi.failuredetector.FailureRetryPolicy;
import java.util.Properties;
public class TimeoutFailureRetryFactory
extends FailureRetryPolicyFactory
{
@Override
public AbstractFailureRetryPolicy getFailureRetryPolicy(Properties properties)
{
return new TimeoutFailureRetryPolicy(new TimeoutFailureRetryConfig(properties));
}
@Override
public AbstractFailureRetryPolicy getFailureRetryPolicy(Properties properties, Ticker ticker)
{
return new TimeoutFailureRetryPolicy(new TimeoutFailureRetryConfig(properties), ticker);
}
@Override
public String getName()
{
return FailureRetryPolicy.TIMEOUT;
}
}

43
presto-main/src/main/java/io/prestosql/failuredetector/TimeoutFailureRetryPolicy.java Normal file
View File

@ -0,0 +1,43 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.failuredetector;
import com.google.common.base.Ticker;
import io.prestosql.server.remotetask.Backoff;
import io.prestosql.spi.HostAddress;
public class TimeoutFailureRetryPolicy
extends AbstractFailureRetryPolicy
{
private TimeoutFailureRetryConfig config;
public TimeoutFailureRetryPolicy(TimeoutFailureRetryConfig config)
{
super(new Backoff(config.getMaxTimeoutDuration()));
this.config = config;
}
public TimeoutFailureRetryPolicy(TimeoutFailureRetryConfig config, Ticker ticker)
{
super(new Backoff(config.getMaxTimeoutDuration(), ticker));
this.config = config;
}
@Override
public boolean hasFailed(HostAddress address)
{
return getBackoff().failure();
}
}

37
presto-main/src/main/java/io/prestosql/operator/ExchangeClient.java
View File

@ -22,11 +22,10 @@ import com.google.common.util.concurrent.SettableFuture;
import io.airlift.http.client.HttpClient;
import io.airlift.log.Logger;
import io.airlift.units.DataSize;
import io.airlift.units.Duration;
import io.hetu.core.transport.execution.buffer.PageCodecMarker;
import io.hetu.core.transport.execution.buffer.PagesSerde;
import io.hetu.core.transport.execution.buffer.SerializedPage;
import io.prestosql.failuredetector.FailureDetector;
import io.prestosql.failuredetector.FailureDetectorManager;
import io.prestosql.memory.context.LocalMemoryContext;
import io.prestosql.operator.HttpPageBufferClient.ClientCallback;
import io.prestosql.operator.WorkProcessor.ProcessState;
@ -74,13 +73,10 @@ public class ExchangeClient
private final long bufferCapacity;
private final DataSize maxResponseSize;
private final int concurrentRequestMultiplier;
private final Duration maxErrorDuration;
private final boolean acknowledgePages;
private final HttpClient httpClient;
private final ScheduledExecutorService scheduler;
private final FailureDetector failureDetector;
private final boolean detectTimeoutFailures;
private final int maxRetryCount;
private final FailureDetectorManager failureDetectorManager;
@GuardedBy("this")
private boolean noMoreLocations;
@ -134,44 +130,24 @@ public class ExchangeClient
// ExchangeClientStatus.mergeWith assumes all clients have the same bufferCapacity.
// Please change that method accordingly when this assumption becomes not true.
public ExchangeClient(DataSize bufferCapacity,
DataSize maxResponseSize,
int concurrentRequestMultiplier,
Duration maxErrorDuration,
boolean acknowledgePages,
HttpClient httpClient,
ScheduledExecutorService scheduler,
LocalMemoryContext systemMemoryContext,
Executor pageBufferClientCallbackExecutor,
FailureDetector failureDetector)
{
this(bufferCapacity, maxResponseSize, concurrentRequestMultiplier, maxErrorDuration, acknowledgePages, httpClient, scheduler, systemMemoryContext, pageBufferClientCallbackExecutor, failureDetector, ExchangeClientConfig.DETECT_TIMEOUT_FAILURES, ExchangeClientConfig.MAX_RETRY_COUNT);
}
public ExchangeClient(DataSize bufferCapacity,
DataSize maxResponseSize,
int concurrentRequestMultiplier,
Duration maxErrorDuration,
boolean acknowledgePages,
HttpClient httpClient,
ScheduledExecutorService scheduler,
LocalMemoryContext systemMemoryContext,
Executor pageBufferClientCallbackExecutor,
FailureDetector failureDetector,
boolean detectTimeoutFailures,
int maxRetryCount)
FailureDetectorManager failureDetectorManager)
{
this.bufferCapacity = bufferCapacity.toBytes();
this.maxResponseSize = maxResponseSize;
this.concurrentRequestMultiplier = concurrentRequestMultiplier;
this.maxErrorDuration = maxErrorDuration;
this.acknowledgePages = acknowledgePages;
this.httpClient = httpClient;
this.scheduler = scheduler;
this.systemMemoryContext = systemMemoryContext;
this.failureDetector = failureDetector;
this.detectTimeoutFailures = detectTimeoutFailures;
this.maxRetryCount = maxRetryCount;
this.failureDetectorManager = failureDetectorManager;
this.maxBufferRetainedSizeInBytes = Long.MIN_VALUE;
this.pageBufferClientCallbackExecutor = requireNonNull(pageBufferClientCallbackExecutor, "pageBufferClientCallbackExecutor is null");
}
@ -270,7 +246,6 @@ public class ExchangeClient
HttpPageBufferClient client = new HttpPageBufferClient(
httpClient,
maxResponseSize,
maxErrorDuration,
acknowledgePages,
location,
new ExchangeClientCallback(uri),
@ -278,9 +253,7 @@ public class ExchangeClient
pageBufferClientCallbackExecutor,
snapshotEnabled,
querySnapshotManager,
failureDetector,
detectTimeoutFailures,
maxRetryCount);
failureDetectorManager);
allClients.put(uri, client);
queuedClients.add(client);

61
presto-main/src/main/java/io/prestosql/operator/ExchangeClientConfig.java
View File

@ -17,29 +17,20 @@ import io.airlift.configuration.Config;
import io.airlift.http.client.HttpClientConfig;
import io.airlift.units.DataSize;
import io.airlift.units.DataSize.Unit;
import io.airlift.units.Duration;
import io.airlift.units.MinDataSize;
import io.airlift.units.MinDuration;
import javax.validation.constraints.Min;
import javax.validation.constraints.NotNull;
import java.util.concurrent.TimeUnit;
public class ExchangeClientConfig
{
public static final boolean DETECT_TIMEOUT_FAILURES = true;
public static final int MAX_RETRY_COUNT = 100;
private DataSize maxBufferSize = new DataSize(32, Unit.MEGABYTE);
private int concurrentRequestMultiplier = 3;
private final Duration minErrorDuration = new Duration(1, TimeUnit.MINUTES);
private Duration maxErrorDuration = new Duration(5, TimeUnit.MINUTES);
private DataSize maxResponseSize = new HttpClientConfig().getMaxContentLength();
private int clientThreads = 25;
private int pageBufferClientMaxCallbackThreads = 25;
private boolean acknowledgePages = true;
private int maxRetryCount = MAX_RETRY_COUNT;
private boolean detectTimeoutFailures = true;
@NotNull
public DataSize getMaxBufferSize()
@ -67,58 +58,6 @@ public class ExchangeClientConfig
return this;
}
@Deprecated
public Duration getMinErrorDuration()
{
return maxErrorDuration;
}
@Deprecated
@Config("exchange.min-error-duration")
public ExchangeClientConfig setMinErrorDuration(Duration minErrorDuration)
{
return this;
}
@NotNull
@MinDuration("1ms")
public Duration getMaxErrorDuration()
{
return maxErrorDuration;
}
@Config("exchange.is-timeout-failure-detection-enabled")
public ExchangeClientConfig setDetectTimeoutFailures(boolean b)
{
this.detectTimeoutFailures = b;
return this;
}
public boolean getDetectTimeoutFailures()
{
return this.detectTimeoutFailures;
}
@Config("exchange.max-error-duration")
public ExchangeClientConfig setMaxErrorDuration(Duration maxErrorDuration)
{
this.maxErrorDuration = maxErrorDuration;
return this;
}
@Config("exchange.max-retry-count")
public ExchangeClientConfig setMaxRetryCount(int maxRetryCount)
{
this.maxRetryCount = maxRetryCount;
return this;
}
@Min(100)
public int getMaxRetryCount()
{
return this.maxRetryCount;
}
@NotNull
@MinDataSize("1MB")
public DataSize getMaxResponseSize()

49
presto-main/src/main/java/io/prestosql/operator/ExchangeClientFactory.java
View File

@ -16,8 +16,7 @@ package io.prestosql.operator;
import io.airlift.concurrent.ThreadPoolExecutorMBean;
import io.airlift.http.client.HttpClient;
import io.airlift.units.DataSize;
import io.airlift.units.Duration;
import io.prestosql.failuredetector.FailureDetector;
import io.prestosql.failuredetector.FailureDetectorManager;
import io.prestosql.memory.context.LocalMemoryContext;
import org.weakref.jmx.Managed;
import org.weakref.jmx.Nested;
@ -40,59 +39,47 @@ public class ExchangeClientFactory
{
private final DataSize maxBufferedBytes;
private final int concurrentRequestMultiplier;
private final Duration maxErrorDuration;
private final HttpClient httpClient;
private final DataSize maxResponseSize;
private final boolean acknowledgePages;
private final ScheduledExecutorService scheduler;
private final ThreadPoolExecutorMBean executorMBean;
private final ExecutorService pageBufferClientCallbackExecutor;
private final FailureDetector failureDetector;
private final boolean detectTimeoutFailures;
private final int maxRetryCount;
private final FailureDetectorManager failureDetectorManager;
@Inject
public ExchangeClientFactory(
ExchangeClientConfig config,
@ForExchange HttpClient httpClient,
@ForExchange ScheduledExecutorService scheduler,
FailureDetector failureDetector)
FailureDetectorManager failureDetectorManager)
{
this(
config.getMaxBufferSize(),
config.getMaxResponseSize(),
config.getConcurrentRequestMultiplier(),
config.getMaxErrorDuration(),
config.isAcknowledgePages(),
config.getPageBufferClientMaxCallbackThreads(),
httpClient,
scheduler,
failureDetector,
config.getDetectTimeoutFailures(),
config.getMaxRetryCount());
failureDetectorManager);
}
public ExchangeClientFactory(
DataSize maxBufferedBytes,
DataSize maxResponseSize,
int concurrentRequestMultiplier,
Duration maxErrorDuration,
boolean acknowledgePages,
int pageBufferClientMaxCallbackThreads,
HttpClient httpClient,
ScheduledExecutorService scheduler,
FailureDetector failureDetector,
boolean detectTimeoutFailures,
int maxRetryCount)
FailureDetectorManager failureDetectorManager)
{
this.maxBufferedBytes = requireNonNull(maxBufferedBytes, "maxBufferedBytes is null");
this.concurrentRequestMultiplier = concurrentRequestMultiplier;
this.maxErrorDuration = requireNonNull(maxErrorDuration, "maxErrorDuration is null");
this.acknowledgePages = acknowledgePages;
this.httpClient = requireNonNull(httpClient, "httpClient is null");
this.failureDetector = requireNonNull(failureDetector, "failureDetector is null");
this.detectTimeoutFailures = detectTimeoutFailures;
this.maxRetryCount = maxRetryCount;
this.failureDetectorManager = failureDetectorManager;
// Use only 0.75 of the maxResponseSize to leave room for additional bytes from the encoding
// TODO figure out a better way to compute the size of data that will be transferred over the network
@ -110,27 +97,6 @@ public class ExchangeClientFactory
checkArgument(concurrentRequestMultiplier > 0, "concurrentRequestMultiplier must be at least 1: %s", concurrentRequestMultiplier);
}
public ExchangeClientFactory(
DataSize maxBufferedBytes,
DataSize maxResponseSize,
int concurrentRequestMultiplier,
Duration maxErrorDuration,
boolean acknowledgePages,
int pageBufferClientMaxCallbackThreads,
HttpClient httpClient,
ScheduledExecutorService scheduler,
FailureDetector failureDetector)
{
this(maxBufferedBytes, maxResponseSize,
concurrentRequestMultiplier,
maxErrorDuration,
acknowledgePages,
pageBufferClientMaxCallbackThreads,
httpClient, scheduler, failureDetector,
ExchangeClientConfig.DETECT_TIMEOUT_FAILURES,
ExchangeClientConfig.MAX_RETRY_COUNT);
}
@PreDestroy
public void stop()
{
@ -151,12 +117,11 @@ public class ExchangeClientFactory
maxBufferedBytes,
maxResponseSize,
concurrentRequestMultiplier,
maxErrorDuration,
acknowledgePages,
httpClient,
scheduler,
systemMemoryContext,
pageBufferClientCallbackExecutor,
failureDetector, detectTimeoutFailures, maxRetryCount);
failureDetectorManager);
}
}

122
presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java
View File

@ -31,12 +31,12 @@ import io.airlift.log.Logger;
import io.airlift.slice.InputStreamSliceInput;
import io.airlift.slice.SliceInput;
import io.airlift.units.DataSize;
import io.airlift.units.Duration;
import io.hetu.core.transport.execution.buffer.SerializedPage;
import io.prestosql.failuredetector.FailureDetector;
import io.prestosql.server.remotetask.Backoff;
import io.prestosql.failuredetector.FailureDetectorManager;
import io.prestosql.snapshot.QuerySnapshotManager;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.failuredetector.FailureRetryPolicy;
import io.prestosql.spi.failuredetector.IBackoff;
import org.joda.time.DateTime;
import javax.annotation.Nullable;
@ -116,8 +116,7 @@ public final class HttpPageBufferClient
private final URI location;
private final ClientCallback clientCallback;
private final ScheduledExecutorService scheduler;
private final Backoff backoff;
private final FailureDetector failureDetector;
private final IBackoff backoff;
@GuardedBy("this")
private boolean closed;
@ -148,42 +147,19 @@ public final class HttpPageBufferClient
private final boolean isSnapshotEnabled;
private final QuerySnapshotManager querySnapshotManager;
private boolean detectTimeoutFailures;
private final FailureRetryPolicy failureRetryPolicy;
public HttpPageBufferClient(
HttpClient httpClient,
DataSize maxResponseSize,
Duration maxErrorDuration,
boolean acknowledgePages,
TaskLocation location,
ClientCallback clientCallback,
ScheduledExecutorService scheduler,
Executor pageBufferClientCallbackExecutor,
boolean isSnapshotEnabled,
QuerySnapshotManager querySnapshotManager,
FailureDetector failureDetector,
boolean detectTimeoutFailures,
int maxRetryCount)
{
this(httpClient, maxResponseSize, maxErrorDuration, acknowledgePages, location, clientCallback, scheduler, Ticker.systemTicker(), pageBufferClientCallbackExecutor, isSnapshotEnabled, querySnapshotManager, failureDetector, detectTimeoutFailures, maxRetryCount);
}
@VisibleForTesting
HttpPageBufferClient(
HttpClient httpClient,
DataSize maxResponseSize,
Duration maxErrorDuration,
boolean acknowledgePages,
TaskLocation location,
ClientCallback clientCallback,
ScheduledExecutorService scheduler,
Ticker ticker,
Executor pageBufferClientCallbackExecutor,
boolean isSnapshotEnabled,
QuerySnapshotManager querySnapshotManager,
FailureDetector failureDetector,
boolean detectTimeoutFailures,
int maxRetryCount)
FailureDetectorManager failureDetectorManager)
{
this.httpClient = requireNonNull(httpClient, "httpClient is null");
this.maxResponseSize = requireNonNull(maxResponseSize, "maxResponseSize is null");
@ -194,13 +170,41 @@ public final class HttpPageBufferClient
this.scheduler = requireNonNull(scheduler, "scheduler is null");
this.taskInstanceId = requireNonNull(location.getInstanceId(), "taskInstanceId is null");
this.pageBufferClientCallbackExecutor = requireNonNull(pageBufferClientCallbackExecutor, "pageBufferClientCallbackExecutor is null");
requireNonNull(maxErrorDuration, "maxErrorDuration is null");
requireNonNull(ticker, "ticker is null");
this.backoff = new Backoff(maxErrorDuration, ticker, maxRetryCount);
requireNonNull(failureDetectorManager, "failure detection manager is null");
this.failureRetryPolicy = failureDetectorManager.getFailureRetryPolicy(failureDetectorManager.getFailureRetryPolicyUserProfile());
this.backoff = this.failureRetryPolicy.getBackoff();
this.isSnapshotEnabled = isSnapshotEnabled;
this.querySnapshotManager = querySnapshotManager;
}
@VisibleForTesting
HttpPageBufferClient(
HttpClient httpClient,
DataSize maxResponseSize,
boolean acknowledgePages,
TaskLocation location,
ClientCallback clientCallback,
ScheduledExecutorService scheduler,
Executor pageBufferClientCallbackExecutor,
boolean isSnapshotEnabled,
QuerySnapshotManager querySnapshotManager,
Ticker ticker,
FailureDetectorManager failureDetectorManager)
{
this.httpClient = requireNonNull(httpClient, "httpClient is null");
this.maxResponseSize = requireNonNull(maxResponseSize, "maxResponseSize is null");
this.acknowledgePages = acknowledgePages;
requireNonNull(location, "TaskLocation is null");
this.location = requireNonNull(location.getUri(), "location is null");
this.clientCallback = requireNonNull(clientCallback, "clientCallback is null");
this.scheduler = requireNonNull(scheduler, "scheduler is null");
this.taskInstanceId = requireNonNull(location.getInstanceId(), "taskInstanceId is null");
this.pageBufferClientCallbackExecutor = requireNonNull(pageBufferClientCallbackExecutor, "pageBufferClientCallbackExecutor is null");
requireNonNull(failureDetectorManager, "failure detection manager is null");
this.failureRetryPolicy = failureDetectorManager.getFailureRetryPolicy(failureDetectorManager.getFailureRetryPolicyUserProfile(), ticker);
this.backoff = this.failureRetryPolicy.getBackoff();
this.isSnapshotEnabled = isSnapshotEnabled;
this.querySnapshotManager = querySnapshotManager;
this.failureDetector = failureDetector;
this.detectTimeoutFailures = detectTimeoutFailures;
}
public synchronized PageBufferClientStatus getStatus()
@ -425,41 +429,23 @@ public final class HttpPageBufferClient
checkNotHoldsLock(this);
Throwable throwable = rewriteException(t);
if (!(throwable instanceof PrestoException)) {
boolean hasFailed;
if (detectTimeoutFailures) {
// timeout based failure detection
hasFailed = backoff.failure();
}
else { // max-retry-count based failure detection
boolean fail = failureRetryPolicy.hasFailed(fromUri(uri));
/**
* if max retry requests failed, check failure detector info on remote host.
* If node state is gone or unresponsive (e.g. GC pause), immediately fail.
* if node is otherwise (e.g.active), keep retrying till timeout of maxErrorDuration
*/
FailureDetector.State state = failureDetector.getState(fromUri(uri));
log.debug("failure detector state is " + state.toString());
hasFailed = (backoff.maxTried() &&
!FailureDetector.State.ALIVE.equals(state)
|| backoff.timeout());
}
if (hasFailed) {
String message = format("%s (%s - %s failures, failure duration %s, total failed request time %s)",
WORKER_NODE_ERROR,
uri,
backoff.getFailureCount(),
backoff.getFailureDuration().convertTo(SECONDS),
backoff.getFailureRequestTimeTotal().convertTo(SECONDS));
if (querySnapshotManager != null) {
// Snapshot: recover from remote server errors
log.debug(throwable, "Snapshot: remote task failed with resumable error: %s", message);
querySnapshotManager.cancelToResume();
handleFailure(throwable, resultFuture);
return;
}
throwable = new PageTransportTimeoutException(fromUri(uri), message, throwable);
if (!(throwable instanceof PrestoException) && fail) {
String message = format("%s (%s - %s failures, failure duration %s, total failed request time %s)",
WORKER_NODE_ERROR,
uri,
backoff.getFailureCount(),
backoff.getFailureDuration().convertTo(SECONDS),
backoff.getFailureRequestTimeTotal().convertTo(SECONDS));
if (querySnapshotManager != null) {
// Snapshot: recover from remote server errors
log.debug(throwable, "Snapshot: remote task failed with resumable error: %s", message);
querySnapshotManager.cancelToResume();
handleFailure(throwable, resultFuture);
return;
}
throwable = new PageTransportTimeoutException(fromUri(uri), message, throwable);
}
handleFailure(throwable, resultFuture);
}

15
presto-main/src/main/java/io/prestosql/server/PluginManager.java
View File

@ -23,6 +23,8 @@ import io.prestosql.connector.ConnectorManager;
import io.prestosql.cube.CubeManager;
import io.prestosql.eventlistener.EventListenerManager;
import io.prestosql.execution.resourcegroups.ResourceGroupManager;
import io.prestosql.failuredetector.FailureDetectorManager;
import io.prestosql.failuredetector.FailureDetectorPlugin;
import io.prestosql.filesystem.FileSystemClientManager;
import io.prestosql.heuristicindex.HeuristicIndexerManager;
import io.prestosql.metadata.MetadataManager;
@ -38,6 +40,7 @@ import io.prestosql.spi.classloader.ThreadContextClassLoader;
import io.prestosql.spi.connector.ConnectorFactory;
import io.prestosql.spi.cube.CubeProvider;
import io.prestosql.spi.eventlistener.EventListenerFactory;
import io.prestosql.spi.failuredetector.FailureRetryFactory;
import io.prestosql.spi.filesystem.HetuFileSystemClientFactory;
import io.prestosql.spi.function.FunctionNamespaceManagerFactory;
import io.prestosql.spi.function.SqlFunction;
@ -110,6 +113,7 @@ public class PluginManager
private final SeedStoreManager seedStoreManager;
private final HetuMetaStoreManager hetuMetaStoreManager;
private final FileSystemClientManager fileSystemClientManager;
private final FailureDetectorManager failureDetectorManager;
private final HeuristicIndexerManager heuristicIndexerManager;
private final SessionPropertyDefaults sessionPropertyDefaults;
private final ArtifactResolver resolver;
@ -137,7 +141,8 @@ public class PluginManager
SeedStoreManager seedStoreManager,
FileSystemClientManager fileSystemClientManager,
HetuMetaStoreManager hetuMetaStoreManager,
HeuristicIndexerManager heuristicIndexerManager)
HeuristicIndexerManager heuristicIndexerManager,
FailureDetectorManager failureDetectorManager)
{
requireNonNull(nodeInfo, "nodeInfo is null");
requireNonNull(config, "config is null");
@ -169,6 +174,7 @@ public class PluginManager
this.fileSystemClientManager = requireNonNull(fileSystemClientManager, "fileSystemClientManager is null");
this.hetuMetaStoreManager = requireNonNull(hetuMetaStoreManager, "hetuMetaStoreManager is null");
this.heuristicIndexerManager = requireNonNull(heuristicIndexerManager, "heuristicIndexerManager is null");
this.failureDetectorManager = requireNonNull(failureDetectorManager, "failureDetectorManager is null");
}
public void loadPlugins()
@ -344,6 +350,13 @@ public class PluginManager
heuristicIndexerManager.loadIndexFactories(indexFactory);
}
// to-do: make failure detector as a plugin
FailureDetectorPlugin fplugin = new FailureDetectorPlugin();
for (FailureRetryFactory failureRetryFactory : fplugin.getFailureRetryFactory()) {
log.info("Registering failure retry policy provider %s", failureRetryFactory.getName());
FailureDetectorManager.addFailureRetryFactory(failureRetryFactory);
}
installFunctionsPlugin(plugin);
}

4
presto-main/src/main/java/io/prestosql/server/PrestoServer.java
View File

@ -42,6 +42,7 @@ import io.prestosql.eventlistener.EventListenerModule;
import io.prestosql.execution.resourcegroups.ResourceGroupManager;
import io.prestosql.execution.scheduler.NodeSchedulerConfig;
import io.prestosql.execution.warnings.WarningCollectorModule;
import io.prestosql.failuredetector.FailureDetectorManager;
import io.prestosql.filesystem.FileSystemClientManager;
import io.prestosql.heuristicindex.HeuristicIndexerManager;
import io.prestosql.httpserver.HetuHttpServerInfo;
@ -145,6 +146,9 @@ public class PrestoServer
FileSystemClientManager fileSystemClientManager = injector.getInstance(FileSystemClientManager.class);
fileSystemClientManager.loadFactoryConfigs();
FailureDetectorManager failureDetectorManager = injector.getInstance(FailureDetectorManager.class);
failureDetectorManager.loadFactoryConfigs();
injector.getInstance(SeedStoreManager.class).loadSeedStore();
if (injector.getInstance(SeedStoreManager.class).isSeedStoreOnYarnEnabled()) {
addSeedOnYarnInformation(

5
presto-main/src/main/java/io/prestosql/server/ServerMainModule.java
View File

@ -70,6 +70,8 @@ import io.prestosql.execution.scheduler.NetworkTopology;
import io.prestosql.execution.scheduler.NodeScheduler;
import io.prestosql.execution.scheduler.NodeSchedulerConfig;
import io.prestosql.execution.scheduler.NodeSchedulerExporter;
import io.prestosql.failuredetector.FailureDetectorManager;
import io.prestosql.failuredetector.FailureRetryConfig;
import io.prestosql.filesystem.FileSystemClientManager;
import io.prestosql.heuristicindex.HeuristicIndexerManager;
import io.prestosql.index.IndexManager;
@ -386,6 +388,9 @@ public class ServerMainModule
jsonCodecBinder(binder).bindJsonCodec(ExecutionFailureInfo.class);
jaxrsBinder(binder).bind(PagesResponseWriter.class);
binder.bind(FailureDetectorManager.class).in(Scopes.SINGLETON);
configBinder(binder).bindConfig(FailureRetryConfig.class);
// exchange client
binder.bind(ExchangeClientSupplier.class).to(ExchangeClientFactory.class).in(Scopes.SINGLETON);
httpClientBinder(binder).bindHttpClient("exchange", ForExchange.class)

112
presto-main/src/main/java/io/prestosql/server/remotetask/Backoff.java
View File

@ -15,10 +15,8 @@ package io.prestosql.server.remotetask;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Ticker;
import com.google.common.collect.ImmutableList;
import io.airlift.log.Logger;
import io.airlift.units.Duration;
import io.prestosql.operator.ExchangeClientConfig;
import io.prestosql.spi.failuredetector.IBackoff;
import javax.annotation.concurrent.ThreadSafe;
@ -33,30 +31,19 @@ import static java.util.concurrent.TimeUnit.NANOSECONDS;
@ThreadSafe
public class Backoff
implements IBackoff
{
private static final Logger log = Logger.get(Backoff.class);
private static final int MIN_RETRIES = 3;
private static final int MAX_RETRIES = ExchangeClientConfig.MAX_RETRY_COUNT;
private static final List<Duration> DEFAULT_BACKOFF_DELAY_INTERVALS = ImmutableList.<Duration>builder()
.add(new Duration(0, MILLISECONDS))
.add(new Duration(50, MILLISECONDS))
.add(new Duration(100, MILLISECONDS))
.add(new Duration(200, MILLISECONDS))
.add(new Duration(500, MILLISECONDS))
.build();
protected final int minTries;
protected final long maxFailureIntervalNanos;
protected final Ticker ticker;
protected final long[] backoffDelayIntervalsNanos;
private final int minTries;
private final int maxTries;
private final long maxFailureIntervalNanos;
private final Ticker ticker;
private final long[] backoffDelayIntervalsNanos;
protected long firstFailureTime;
protected long lastFailureTime;
protected long failureCount;
protected long failureRequestTimeTotal;
private long firstFailureTime;
private long lastFailureTime;
private long failureCount;
private long failureRequestTimeTotal;
private long lastRequestStart;
protected long lastRequestStart;
public Backoff(Duration maxFailureInterval)
{
@ -65,30 +52,7 @@ public class Backoff
public Backoff(Duration maxFailureInterval, Ticker ticker)
{
this(MIN_RETRIES, maxFailureInterval, ticker, DEFAULT_BACKOFF_DELAY_INTERVALS, MAX_RETRIES);
}
public Backoff(Duration maxFailureInterval, Ticker ticker, int maxTries)
{
this(MIN_RETRIES, maxFailureInterval, ticker, DEFAULT_BACKOFF_DELAY_INTERVALS, maxTries);
}
@VisibleForTesting
public Backoff(int minTries, Duration maxFailureInterval, Ticker ticker, List<Duration> backoffDelayIntervals, int maxTries)
{
checkArgument(minTries > 0, "minTries must be at least 1");
requireNonNull(maxFailureInterval, "maxFailureInterval is null");
requireNonNull(ticker, "ticker is null");
requireNonNull(backoffDelayIntervals, "backoffDelayIntervals is null");
checkArgument(!backoffDelayIntervals.isEmpty(), "backoffDelayIntervals must contain at least one entry");
this.minTries = minTries;
this.maxTries = (MAX_RETRIES < maxTries) ? maxTries : MAX_RETRIES;
this.maxFailureIntervalNanos = maxFailureInterval.roundTo(NANOSECONDS);
this.ticker = ticker;
this.backoffDelayIntervalsNanos = backoffDelayIntervals.stream()
.mapToLong(duration -> duration.roundTo(NANOSECONDS))
.toArray();
this(MIN_RETRIES, maxFailureInterval, ticker, DEFAULT_BACKOFF_DELAY_INTERVALS);
}
@VisibleForTesting
@ -101,7 +65,6 @@ public class Backoff
checkArgument(!backoffDelayIntervals.isEmpty(), "backoffDelayIntervals must contain at least one entry");
this.minTries = minTries;
this.maxTries = (minTries < MAX_RETRIES) ? MAX_RETRIES : minTries;
this.maxFailureIntervalNanos = maxFailureInterval.roundTo(NANOSECONDS);
this.ticker = ticker;
this.backoffDelayIntervalsNanos = backoffDelayIntervals.stream()
@ -137,57 +100,18 @@ public class Backoff
{
lastRequestStart = 0;
firstFailureTime = 0;
setFailureCount(0, false);
resetFailureCount();
lastFailureTime = 0;
}
private synchronized void setFailureCount(int n, boolean isInc)
protected synchronized void resetFailureCount()
{
if (isInc) {
failureCount = failureCount + n;
return;
}
failureCount = n;
failureCount = 0;
}
/**
* @return true if max retry failed, now it is time to check node status from HeartbeatFailureDetector
*/
public synchronized boolean maxTried()
protected synchronized void updateFailureCount()
{
long now = ticker.read();
lastFailureTime = now;
setFailureCount(1, true);
if (lastRequestStart != 0) {
failureRequestTimeTotal += now - lastRequestStart;
lastRequestStart = 0;
}
if (firstFailureTime == 0) {
firstFailureTime = now;
// can not fail on first failure
return false;
}
if (getFailureCount() < minTries) {
return false;
}
if (getFailureCount() >= maxTries) {
log.debug("failure retry count cross max retry number " + maxTries);
}
return getFailureCount() >= maxTries;
}
/**
* @return true if maxErrorDuration is passed. Does not matter how many retry has happened.
*/
public synchronized boolean timeout()
{
long now = ticker.read();
long failureDuration = now - firstFailureTime;
return failureDuration >= maxFailureIntervalNanos;
failureCount++;
}
/**
@ -199,7 +123,7 @@ public class Backoff
long now = ticker.read();
lastFailureTime = now;
setFailureCount(1, true);
updateFailureCount();
if (lastRequestStart != 0) {
failureRequestTimeTotal += now - lastRequestStart;
lastRequestStart = 0;

98
presto-main/src/main/java/io/prestosql/server/remotetask/MaxRetryBackoff.java Normal file
View File

@ -0,0 +1,98 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.server.remotetask;
import com.google.common.base.Ticker;
import io.airlift.log.Logger;
import io.airlift.units.Duration;
import io.prestosql.spi.failuredetector.FailureRetryPolicy;
import io.prestosql.spi.failuredetector.IBackoff;
public class MaxRetryBackoff
extends Backoff
implements IBackoff
{
private static final Logger log = Logger.get(MaxRetryBackoff.class);
private final int maxTries;
MaxRetryBackoff(Duration maxFailureInterval, int maxTries)
{
super(maxFailureInterval);
this.maxTries = (minTries < maxTries) ? maxTries : minTries;
}
public MaxRetryBackoff(int maxTries)
{
this(Duration.valueOf(FailureRetryPolicy.DEFAULT_TIMEOUT_DURATION), maxTries);
}
public MaxRetryBackoff(int maxTries, Ticker ticker)
{
this(Duration.valueOf(FailureRetryPolicy.DEFAULT_TIMEOUT_DURATION), maxTries, ticker);
}
public MaxRetryBackoff()
{
this(Duration.valueOf(FailureRetryPolicy.DEFAULT_TIMEOUT_DURATION), Integer.parseInt(FailureRetryPolicy.MAX_RETRY_COUNT));
}
public MaxRetryBackoff(Duration maxFailureInterval, int maxTries, Ticker ticker)
{
super(maxFailureInterval, ticker);
this.maxTries = (minTries < maxTries) ? maxTries : minTries;
}
/**
* @return true if max retry failed, now it is time to check node status from HeartbeatFailureDetector
*/
public synchronized boolean maxRetryDone()
{
long now = ticker.read();
lastFailureTime = now;
updateFailureCount();
if (lastRequestStart != 0) {
failureRequestTimeTotal += now - lastRequestStart;
lastRequestStart = 0;
}
if (firstFailureTime == 0) {
firstFailureTime = now;
// can not fail on first failure
return false;
}
if (getFailureCount() < minTries) {
return false;
}
log.debug(" failure count " + getFailureCount());
if (getFailureCount() >= maxTries) {
log.debug(" failure count has crossed max retry count " + maxTries);
}
return getFailureCount() >= maxTries;
}
/**
* @return true if maxErrorDuration is passed. Does not matter how many retry has happened.
*/
public synchronized boolean timeout()
{
long now = ticker.read();
long failureDuration = now - firstFailureTime;
return failureDuration >= maxFailureIntervalNanos;
}
}

20
presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java
View File

@ -80,6 +80,10 @@ import io.prestosql.execution.scheduler.LegacyNetworkTopology;
import io.prestosql.execution.scheduler.NodeScheduler;
import io.prestosql.execution.scheduler.NodeSchedulerConfig;
import io.prestosql.execution.warnings.WarningCollector;
import io.prestosql.failuredetector.FailureDetectorManager;
import io.prestosql.failuredetector.FailureRetryConfig;
import io.prestosql.failuredetector.NoOpFailureDetector;
import io.prestosql.failuredetector.TimeoutFailureRetryFactory;
import io.prestosql.filesystem.FileSystemClientManager;
import io.prestosql.heuristicindex.HeuristicIndexerManager;
import io.prestosql.index.IndexManager;
@ -124,6 +128,7 @@ import io.prestosql.spi.Plugin;
import io.prestosql.spi.connector.CatalogName;
import io.prestosql.spi.connector.ConnectorFactory;
import io.prestosql.spi.connector.QualifiedObjectName;
import io.prestosql.spi.failuredetector.FailureRetryPolicy;
import io.prestosql.spi.metadata.TableHandle;
import io.prestosql.spi.operator.ReuseExchangeOperator;
import io.prestosql.spi.plan.PlanNode;
@ -207,6 +212,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@ -417,6 +423,17 @@ public class LocalQueryRunner
SeedStoreManager seedStoreManager = new SeedStoreManager(fileSystemClientManager);
HttpServerConfig httpServerConfig = new HttpServerConfig();
httpServerConfig.setHttpEnabled(false);
FailureRetryConfig cfg = new FailureRetryConfig();
cfg.setFailureRetryPolicyProfile("test");
Properties prop = new Properties();
prop.setProperty(FailureRetryPolicy.FD_RETRY_TYPE, cfg.getFailureRetryPolicyProfile());
prop.setProperty(FailureRetryPolicy.MAX_RETRY_COUNT, "10");
prop.setProperty(FailureRetryPolicy.MAX_TIMEOUT_DURATION, "60s");
FailureDetectorManager.addFrConfigs(cfg.getFailureRetryPolicyProfile(), prop);
FailureDetectorManager.addFailureRetryFactory(new TimeoutFailureRetryFactory());
FailureDetectorManager failureDetectorManager = new FailureDetectorManager(cfg, new NoOpFailureDetector());
this.pluginManager = new PluginManager(
nodeInfo,
new PluginManagerConfig(),
@ -437,7 +454,8 @@ public class LocalQueryRunner
seedStoreManager,
fileSystemClientManager,
hetuMetaStoreManager,
heuristicIndexerManager);
heuristicIndexerManager,
failureDetectorManager);
connectorManager.addConnectorFactory(globalSystemConnectorFactory);
connectorManager.createConnection(GlobalSystemConnector.NAME, GlobalSystemConnector.NAME, ImmutableMap.of());

107
presto-main/src/test/java/io/prestosql/failuredetector/TestFailureDetectionManager.java Normal file
View File

@ -0,0 +1,107 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.failuredetector;
import io.prestosql.spi.failuredetector.FailureRetryPolicy;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
import java.util.Map;
import java.util.Properties;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
public class TestFailureDetectionManager
{
private FailureRetryConfig cfg;
private FailureRetryConfig cfg1;
private FailureDetectorManager failureDetectorManager;
private FailureDetectorManager failureDetectorManager1;
private FailureDetectorManager failureDetectorManager2;
@BeforeClass
public void setUp()
{
cfg = new FailureRetryConfig();
cfg.setFailureRetryPolicyProfile("test2");
Properties prop = new Properties();
prop.setProperty(FailureRetryPolicy.FD_RETRY_TYPE, FailureRetryPolicy.MAXRETRY);
prop.setProperty(FailureRetryPolicy.MAX_RETRY_COUNT, "10");
prop.setProperty(FailureRetryPolicy.MAX_TIMEOUT_DURATION, "60s");
FailureDetectorManager.addFrConfigs(cfg.getFailureRetryPolicyProfile(), prop);
FailureDetectorManager.addFailureRetryFactory(new MaxRetryFailureRetryFactory());
failureDetectorManager = new FailureDetectorManager(cfg, new NoOpFailureDetector());
cfg1 = new FailureRetryConfig();
cfg1.setFailureRetryPolicyProfile("default1");
Properties prop1 = new Properties();
prop1.setProperty(FailureRetryPolicy.FD_RETRY_TYPE, FailureRetryPolicy.MAXRETRY);
prop1.setProperty(FailureRetryPolicy.MAX_RETRY_COUNT, "100");
prop1.setProperty(FailureRetryPolicy.MAX_TIMEOUT_DURATION, "34560s");
FailureDetectorManager.addFrConfigs(cfg1.getFailureRetryPolicyProfile(), prop1);
failureDetectorManager1 = new FailureDetectorManager(cfg1, new NoOpFailureDetector());
failureDetectorManager2 = new FailureDetectorManager(new NoOpFailureDetector(), "30s");
}
@AfterClass(alwaysRun = true)
public void tearDown()
{
FailureDetectorManager.removeallFrConfigs();
}
@Test
public void testProfileConfigsCount()
{
Map<String, Properties> frConfigs = FailureDetectorManager.getAvailableFrConfigs();
assertNotNull(frConfigs.get("default1"));
assertNotNull(frConfigs.get("test2"));
assertNotNull(frConfigs.get("default"));
}
@Test
public void testDefaultRetryProfile()
{
Map<String, Properties> cfgProp = FailureDetectorManager.getAvailableFrConfigs();
Properties checkprop = cfgProp.get("test2");
assertEquals(FailureRetryPolicy.MAXRETRY, checkprop.getProperty(FailureRetryPolicy.FD_RETRY_TYPE));
assertEquals("10", checkprop.getProperty(FailureRetryPolicy.MAX_RETRY_COUNT));
assertEquals("60s", checkprop.getProperty(FailureRetryPolicy.MAX_TIMEOUT_DURATION));
}
@Test
public void testDefaultRetryProfile1()
{
Map<String, Properties> cfgProp = FailureDetectorManager.getAvailableFrConfigs();
Properties checkprop = cfgProp.get("default1");
assertEquals(FailureRetryPolicy.MAXRETRY, checkprop.getProperty(FailureRetryPolicy.FD_RETRY_TYPE));
assertEquals("100", checkprop.getProperty(FailureRetryPolicy.MAX_RETRY_COUNT));
assertEquals("34560s", checkprop.getProperty(FailureRetryPolicy.MAX_TIMEOUT_DURATION));
}
@Test
public void testDefaultRetryProfile2()
{
Map<String, Properties> cfgProp = FailureDetectorManager.getAvailableFrConfigs();
Properties checkprop = cfgProp.get(failureDetectorManager2.getFailureRetryPolicyUserProfile());
assertEquals("default", failureDetectorManager2.getFailureRetryPolicyUserProfile());
assertEquals(FailureRetryPolicy.TIMEOUT, checkprop.getProperty(FailureRetryPolicy.FD_RETRY_TYPE));
assertEquals("30s", checkprop.getProperty(FailureRetryPolicy.MAX_TIMEOUT_DURATION));
}
}

23
presto-main/src/test/java/io/prestosql/operator/TestExchangeClient.java
View File

@ -25,7 +25,9 @@ import io.airlift.units.Duration;
import io.hetu.core.transport.execution.buffer.PagesSerde;
import io.hetu.core.transport.execution.buffer.SerializedPage;
import io.prestosql.block.BlockAssertions;
import io.prestosql.failuredetector.FailureDetectorManager;
import io.prestosql.failuredetector.NoOpFailureDetector;
import io.prestosql.failuredetector.TimeoutFailureRetryFactory;
import io.prestosql.memory.context.SimpleLocalMemoryContext;
import io.prestosql.spi.Page;
import io.prestosql.spi.QueryId;
@ -65,6 +67,7 @@ public class TestExchangeClient
{
private ScheduledExecutorService scheduler;
private ExecutorService pageBufferClientCallbackExecutor;
private FailureDetectorManager failureDetectorManager;
private static final PagesSerde PAGES_SERDE = testingPagesSerde();
@ -73,6 +76,8 @@ public class TestExchangeClient
{
scheduler = newScheduledThreadPool(4, daemonThreadsNamed("test-%s"));
pageBufferClientCallbackExecutor = Executors.newSingleThreadExecutor();
failureDetectorManager = new FailureDetectorManager(new NoOpFailureDetector(), "60s");
FailureDetectorManager.addFailureRetryFactory(new TimeoutFailureRetryFactory());
}
@AfterClass(alwaysRun = true)
@ -106,12 +111,11 @@ public class TestExchangeClient
new DataSize(32, Unit.MEGABYTE),
maxResponseSize,
1,
new Duration(1, TimeUnit.MINUTES),
true,
new TestingHttpClient(processor, scheduler),
scheduler,
new SimpleLocalMemoryContext(newSimpleAggregatedMemoryContext(), "test"),
pageBufferClientCallbackExecutor, new NoOpFailureDetector());
pageBufferClientCallbackExecutor, failureDetectorManager);
exchangeClient.addLocation(new TaskLocation(location, instanceId));
exchangeClient.noMoreLocations();
@ -152,12 +156,11 @@ public class TestExchangeClient
new DataSize(32, Unit.MEGABYTE),
maxResponseSize,
1,
new Duration(1, TimeUnit.MINUTES),
true,
new TestingHttpClient(processor, scheduler),
scheduler,
new SimpleLocalMemoryContext(newSimpleAggregatedMemoryContext(), "test"),
pageBufferClientCallbackExecutor, new NoOpFailureDetector());
pageBufferClientCallbackExecutor, failureDetectorManager);
exchangeClient.setSnapshotEnabled(NOOP_SNAPSHOT_UTILS.getQuerySnapshotManager(new QueryId("query")));
final String target1 = "target1";
@ -207,12 +210,11 @@ public class TestExchangeClient
new DataSize(32, Unit.MEGABYTE),
new DataSize(10, Unit.MEGABYTE),
1,
new Duration(1, TimeUnit.MINUTES),
true,
mock(HttpClient.class),
scheduler,
new SimpleLocalMemoryContext(newSimpleAggregatedMemoryContext(), "test"),
pageBufferClientCallbackExecutor, new NoOpFailureDetector());
pageBufferClientCallbackExecutor, failureDetectorManager);
exchangeClient.setSnapshotEnabled(NOOP_SNAPSHOT_UTILS.getQuerySnapshotManager(new QueryId("query")));
String origin1 = "location1";
@ -247,12 +249,11 @@ public class TestExchangeClient
new DataSize(32, Unit.MEGABYTE),
maxResponseSize,
1,
new Duration(1, TimeUnit.MINUTES),
true,
new TestingHttpClient(processor, newCachedThreadPool(daemonThreadsNamed("test-%s"))),
scheduler,
new SimpleLocalMemoryContext(newSimpleAggregatedMemoryContext(), "test"),
pageBufferClientCallbackExecutor, new NoOpFailureDetector());
pageBufferClientCallbackExecutor, failureDetectorManager);
URI location1 = URI.create("http://localhost:8081/foo");
String instanceId1 = "testing instance id";
@ -322,12 +323,11 @@ public class TestExchangeClient
new DataSize(1, Unit.BYTE),
maxResponseSize,
1,
new Duration(1, TimeUnit.MINUTES),
true,
new TestingHttpClient(processor, newCachedThreadPool(daemonThreadsNamed("test-%s"))),
scheduler,
new SimpleLocalMemoryContext(newSimpleAggregatedMemoryContext(), "test"),
pageBufferClientCallbackExecutor, new NoOpFailureDetector());
pageBufferClientCallbackExecutor, failureDetectorManager);
exchangeClient.addLocation(new TaskLocation(location, instanceId));
exchangeClient.noMoreLocations();
@ -405,12 +405,11 @@ public class TestExchangeClient
new DataSize(1, Unit.BYTE),
maxResponseSize,
1,
new Duration(1, TimeUnit.MINUTES),
true,
new TestingHttpClient(processor, newCachedThreadPool(daemonThreadsNamed("test-%s"))),
scheduler,
new SimpleLocalMemoryContext(newSimpleAggregatedMemoryContext(), "test"),
pageBufferClientCallbackExecutor, new NoOpFailureDetector());
pageBufferClientCallbackExecutor, failureDetectorManager);
exchangeClient.addLocation(new TaskLocation(location, instanceId));
exchangeClient.noMoreLocations();

18
presto-main/src/test/java/io/prestosql/operator/TestExchangeClientConfig.java
View File

@ -16,11 +16,9 @@ package io.prestosql.operator;
import com.google.common.collect.ImmutableMap;
import io.airlift.http.client.HttpClientConfig;
import io.airlift.units.DataSize;
import io.airlift.units.Duration;
import org.testng.annotations.Test;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import static io.airlift.configuration.testing.ConfigAssertions.assertFullMapping;
import static io.airlift.configuration.testing.ConfigAssertions.assertRecordedDefaults;
@ -35,14 +33,10 @@ public class TestExchangeClientConfig
assertRecordedDefaults(recordDefaults(ExchangeClientConfig.class)
.setMaxBufferSize(new DataSize(32, Unit.MEGABYTE))
.setConcurrentRequestMultiplier(3)
.setMinErrorDuration(new Duration(5, TimeUnit.MINUTES))
.setMaxErrorDuration(new Duration(5, TimeUnit.MINUTES))
.setMaxResponseSize(new HttpClientConfig().getMaxContentLength())
.setPageBufferClientMaxCallbackThreads(25)
.setClientThreads(25)
.setAcknowledgePages(true)
.setDetectTimeoutFailures(true)
.setMaxRetryCount(100));
.setAcknowledgePages(true));
}
@Test
@ -51,27 +45,19 @@ public class TestExchangeClientConfig
Map<String, String> properties = new ImmutableMap.Builder<String, String>()
.put("exchange.max-buffer-size", "1GB")
.put("exchange.concurrent-request-multiplier", "13")
.put("exchange.min-error-duration", "13s")
.put("exchange.max-error-duration", "33s")
.put("exchange.max-response-size", "1MB")
.put("exchange.client-threads", "2")
.put("exchange.page-buffer-client.max-callback-threads", "16")
.put("exchange.acknowledge-pages", "false")
.put("exchange.max-retry-count", "110")
.put("exchange.is-timeout-failure-detection-enabled", "false")
.build();
ExchangeClientConfig expected = new ExchangeClientConfig()
.setMaxBufferSize(new DataSize(1, Unit.GIGABYTE))
.setConcurrentRequestMultiplier(13)
.setMinErrorDuration(new Duration(33, TimeUnit.SECONDS))
.setMaxErrorDuration(new Duration(33, TimeUnit.SECONDS))
.setMaxResponseSize(new DataSize(1, Unit.MEGABYTE))
.setClientThreads(2)
.setPageBufferClientMaxCallbackThreads(16)
.setAcknowledgePages(false)
.setMaxRetryCount(110)
.setDetectTimeoutFailures(false);
.setAcknowledgePages(false);
assertFullMapping(properties, expected);
}

7
presto-main/src/test/java/io/prestosql/operator/TestExchangeOperator.java
View File

@ -20,10 +20,11 @@ import com.google.common.collect.ImmutableList;
import io.airlift.http.client.HttpClient;
import io.airlift.http.client.testing.TestingHttpClient;
import io.airlift.units.DataSize;
import io.airlift.units.Duration;
import io.prestosql.Session;
import io.prestosql.execution.Lifespan;
import io.prestosql.failuredetector.FailureDetectorManager;
import io.prestosql.failuredetector.NoOpFailureDetector;
import io.prestosql.failuredetector.TimeoutFailureRetryFactory;
import io.prestosql.metadata.Split;
import io.prestosql.operator.ExchangeOperator.ExchangeOperatorFactory;
import io.prestosql.spi.Page;
@ -87,17 +88,17 @@ public class TestExchangeOperator
scheduledExecutor = newScheduledThreadPool(2, daemonThreadsNamed("test-scheduledExecutor-%s"));
pageBufferClientCallbackExecutor = Executors.newSingleThreadExecutor();
httpClient = new TestingHttpClient(new TestingExchangeHttpClientHandler(taskBuffers), scheduler);
FailureDetectorManager.addFailureRetryFactory(new TimeoutFailureRetryFactory());
exchangeClientSupplier = (systemMemoryUsageListener) -> new ExchangeClient(
new DataSize(32, MEGABYTE),
new DataSize(10, MEGABYTE),
3,
new Duration(1, TimeUnit.MINUTES),
true,
httpClient,
scheduler,
systemMemoryUsageListener,
pageBufferClientCallbackExecutor, new NoOpFailureDetector(), true, 3);
pageBufferClientCallbackExecutor, new FailureDetectorManager(new NoOpFailureDetector(), "60s"));
}
@AfterClass(alwaysRun = true)

134
presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java
View File

@ -21,6 +21,7 @@ import io.airlift.http.client.Request;
import io.airlift.http.client.Response;
import io.airlift.http.client.testing.TestingHttpClient;
import io.airlift.http.client.testing.TestingResponse;
import io.airlift.log.Logger;
import io.airlift.testing.TestingTicker;
import io.airlift.units.DataSize;
import io.airlift.units.DataSize.Unit;
@ -28,10 +29,17 @@ import io.airlift.units.Duration;
import io.hetu.core.transport.execution.buffer.PagesSerde;
import io.hetu.core.transport.execution.buffer.SerializedPage;
import io.prestosql.failuredetector.FailureDetector;
import io.prestosql.failuredetector.FailureDetectorManager;
import io.prestosql.failuredetector.FailureRetryConfig;
import io.prestosql.failuredetector.MaxRetryFailureRetryFactory;
import io.prestosql.failuredetector.MaxRetryFailureRetryPolicy;
import io.prestosql.failuredetector.NoOpFailureDetector;
import io.prestosql.failuredetector.TimeoutFailureRetryFactory;
import io.prestosql.operator.HttpPageBufferClient.ClientCallback;
import io.prestosql.spi.HostAddress;
import io.prestosql.spi.Page;
import io.prestosql.spi.failuredetector.FailureRetryFactory;
import io.prestosql.spi.failuredetector.FailureRetryPolicy;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@ -41,6 +49,8 @@ import java.security.GeneralSecurityException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
@ -65,11 +75,21 @@ import static io.prestosql.testing.TestingPagesSerdeFactory.testingPagesSerde;
import static io.prestosql.util.Failures.WORKER_NODE_ERROR;
import static java.util.concurrent.Executors.newScheduledThreadPool;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.fail;
public class TestHttpPageBufferClient
{
private static final Logger log = Logger.get(TestHttpPageBufferClient.class);
private ScheduledExecutorService scheduler;
private ExecutorService pageBufferClientCallbackExecutor;
private FailureDetectorManager failureDetectorManager;
private FailureDetectorManager failureDetectorManager1;
private FailureDetectorManager failureDetectorManager2;
private FailureDetectorManager failureDetectorManager3;
private FailureRetryConfig cfg;
private FailureRetryPolicy policy;
private static final PagesSerde PAGES_SERDE = testingPagesSerde();
@ -78,6 +98,59 @@ public class TestHttpPageBufferClient
{
scheduler = newScheduledThreadPool(4, daemonThreadsNamed("test-%s"));
pageBufferClientCallbackExecutor = Executors.newSingleThreadExecutor();
cfg = new FailureRetryConfig();
cfg.setFailureRetryPolicyProfile("test1");
Properties prop = new Properties();
prop.setProperty(FailureRetryPolicy.FD_RETRY_TYPE, FailureRetryPolicy.MAXRETRY);
prop.setProperty(FailureRetryPolicy.MAX_RETRY_COUNT, "10");
prop.setProperty(FailureRetryPolicy.MAX_TIMEOUT_DURATION, "300s");
FailureDetectorManager.addFrConfigs(cfg.getFailureRetryPolicyProfile(), prop);
FailureDetectorManager.addFailureRetryFactory(new MaxRetryFailureRetryFactory());
FailureDetectorManager.addFailureRetryFactory(new TimeoutFailureRetryFactory());
failureDetectorManager = new FailureDetectorManager(new NoOpFailureDetector(), "60s");
failureDetectorManager1 = new FailureDetectorManager(cfg, new MockActiveFailureDetector());
failureDetectorManager2 = new FailureDetectorManager(new NoOpFailureDetector(), "30s");
failureDetectorManager3 = new FailureDetectorManager(cfg, new MockNodeCrashFailureDetector());
printGfConfigs();
log.info("searching for " + failureDetectorManager1.getFailureRetryPolicyUserProfile() + " profile");
printRetryFactories();
policy = failureDetectorManager1.getFailureRetryPolicy(failureDetectorManager1.getFailureRetryPolicyUserProfile());
}
@Test
public void TestFaliureRetryPlocies()
{
Map<String, Properties> frConfigs = FailureDetectorManager.getAvailableFrConfigs();
assertNotNull(frConfigs.get("default"));
assertNotNull(frConfigs.get("test1"));
}
private void printRetryFactories()
{
log.info("the following factories are present...");
System.out.println("the following factories are present...");
Map<String, FailureRetryFactory> factories = FailureDetectorManager.getFailureRetryFactories();
for (Map.Entry<String, FailureRetryFactory> e : factories.entrySet()) {
log.info(e.getKey());
}
}
private void printGfConfigs()
{
log.info("the following profiles are present in the factory");
Map<String, Properties> frConfigs = FailureDetectorManager.getAvailableFrConfigs();
for (Map.Entry<String, Properties> e : frConfigs.entrySet()) {
log.info("Profile Name: " + e.getKey());
Properties prop = e.getValue();
log.info(FailureRetryPolicy.FD_RETRY_TYPE + ": " + prop.getProperty(FailureRetryPolicy.FD_RETRY_TYPE));
log.info(FailureRetryPolicy.MAX_RETRY_COUNT + ": " + prop.getProperty(FailureRetryPolicy.MAX_RETRY_COUNT));
log.info(FailureRetryPolicy.MAX_TIMEOUT_DURATION + ": " + prop.getProperty(FailureRetryPolicy.MAX_TIMEOUT_DURATION));
}
}
@AfterClass(alwaysRun = true)
@ -91,6 +164,7 @@ public class TestHttpPageBufferClient
pageBufferClientCallbackExecutor.shutdownNow();
pageBufferClientCallbackExecutor = null;
}
FailureDetectorManager.removeallFrConfigs();
}
@Test
@ -110,14 +184,13 @@ public class TestHttpPageBufferClient
String instanceId = "testing instance id";
HttpPageBufferClient client = new HttpPageBufferClient(new TestingHttpClient(processor, scheduler),
expectedMaxSize,
new Duration(1, TimeUnit.MINUTES),
true,
new TaskLocation(location, instanceId),
callback,
scheduler,
pageBufferClientCallbackExecutor,
false,
null, new NoOpFailureDetector(), false, 10);
null, failureDetectorManager);
assertStatus(client, location, "queued", 0, 0, 0, 0, "not scheduled");
@ -198,14 +271,13 @@ public class TestHttpPageBufferClient
String instanceId = "testing instance id";
HttpPageBufferClient client = new HttpPageBufferClient(new TestingHttpClient(processor, scheduler),
new DataSize(10, Unit.MEGABYTE),
new Duration(1, TimeUnit.MINUTES),
true,
new TaskLocation(location, instanceId),
callback,
scheduler,
pageBufferClientCallbackExecutor,
false,
null, new NoOpFailureDetector(), false, 10);
null, failureDetectorManager);
assertStatus(client, location, "queued", 0, 0, 0, 0, "not scheduled");
@ -241,14 +313,13 @@ public class TestHttpPageBufferClient
String instanceId = "testing instance id";
HttpPageBufferClient client = new HttpPageBufferClient(new TestingHttpClient(processor, scheduler),
new DataSize(10, Unit.MEGABYTE),
new Duration(1, TimeUnit.MINUTES),
true,
new TaskLocation(location, instanceId),
callback,
scheduler,
pageBufferClientCallbackExecutor,
false,
null, new NoOpFailureDetector(), false, 10);
null, failureDetectorManager);
assertStatus(client, location, "queued", 0, 0, 0, 0, "not scheduled");
@ -313,14 +384,13 @@ public class TestHttpPageBufferClient
String instanceId = "testing instance id";
HttpPageBufferClient client = new HttpPageBufferClient(new TestingHttpClient(processor, scheduler),
new DataSize(10, Unit.MEGABYTE),
new Duration(1, TimeUnit.MINUTES),
true,
new TaskLocation(location, instanceId),
callback,
scheduler,
pageBufferClientCallbackExecutor,
false,
null, new NoOpFailureDetector(), false, 10);
null, failureDetectorManager);
assertStatus(client, location, "queued", 0, 0, 0, 0, "not scheduled");
@ -337,6 +407,7 @@ public class TestHttpPageBufferClient
}
catch (BrokenBarrierException ignored) {
// the exception could be ignored
System.out.println(" Exception Ignore");
}
try {
afterRequest.await(10, TimeUnit.SECONDS);
@ -371,15 +442,13 @@ public class TestHttpPageBufferClient
String instanceId = "testing instance id";
HttpPageBufferClient client = new HttpPageBufferClient(new TestingHttpClient(processor, scheduler),
new DataSize(10, Unit.MEGABYTE),
new Duration(30, TimeUnit.SECONDS),
true,
new TaskLocation(location, instanceId),
callback,
scheduler,
ticker,
pageBufferClientCallbackExecutor,
false,
null, new NoOpFailureDetector(), true, 10);
null, ticker, failureDetectorManager2);
assertStatus(client, location, "queued", 0, 0, 0, 0, "not scheduled");
@ -436,34 +505,36 @@ public class TestHttpPageBufferClient
CyclicBarrier requestComplete = new CyclicBarrier(2);
TestingClientCallback callback = new TestingClientCallback(requestComplete);
assertEquals("test1", failureDetectorManager3.getFailureRetryPolicyUserProfile());
if (!(policy instanceof MaxRetryFailureRetryPolicy)) {
fail("faliure detector policy error");
}
URI location = URI.create("http://localhost:8080");
String instanceId = "testing instance id";
HttpPageBufferClient client = new HttpPageBufferClient(new TestingHttpClient(processor, scheduler),
new DataSize(10, Unit.MEGABYTE),
new Duration(300, TimeUnit.SECONDS),
true,
new TaskLocation(location, instanceId),
callback,
scheduler,
ticker,
pageBufferClientCallbackExecutor,
false,
null, new MockNodeCrashFailureDetector(), false, 100);
null, ticker, failureDetectorManager3);
assertStatus(client, location, "queued", 0, 0, 0, 0, "not scheduled");
for (int i = 0; i < 101; i++) {
for (int i = 0; i < 11; i++) {
client.scheduleRequest();
requestComplete.await(10, TimeUnit.SECONDS);
tickerIncrement.set(new Duration(1, TimeUnit.SECONDS));
tickerIncrement.set(new Duration(10, TimeUnit.SECONDS));
}
assertEquals(callback.getPages().size(), 0);
assertEquals(callback.getCompletedRequests(), 101);
assertEquals(callback.getCompletedRequests(), 11);
assertEquals(callback.getFinishedBuffers(), 0);
assertEquals(callback.getFailedBuffers(), 2);
assertInstanceOf(callback.getFailure(), PageTransportTimeoutException.class);
assertContains(callback.getFailure().getMessage(), WORKER_NODE_ERROR + " (http://localhost:8080/0 - 100 failures,");
assertStatus(client, location, "queued", 0, 101, 101, 101, "not scheduled");
assertContains(callback.getFailure().getMessage(), WORKER_NODE_ERROR + " (http://localhost:8080/0 - 10 failures,");
assertStatus(client, location, "queued", 0, 11, 11, 11, "not scheduled");
}
@Test
@ -482,43 +553,46 @@ public class TestHttpPageBufferClient
CyclicBarrier requestComplete = new CyclicBarrier(2);
TestingClientCallback callback = new TestingClientCallback(requestComplete);
if (!(policy instanceof MaxRetryFailureRetryPolicy)) {
fail("faliure detector policy error");
}
assertEquals("test1", failureDetectorManager1.getFailureRetryPolicyUserProfile());
URI location = URI.create("http://localhost:8080");
String instanceId = "testing instance id";
HttpPageBufferClient client = new HttpPageBufferClient(new TestingHttpClient(processor, scheduler),
new DataSize(10, Unit.MEGABYTE),
new Duration(300, TimeUnit.SECONDS),
true,
new TaskLocation(location, instanceId),
callback,
scheduler,
ticker,
pageBufferClientCallbackExecutor,
false,
null, new MockActiveFailureDetector(), false, 100);
null, ticker, failureDetectorManager1);
assertStatus(client, location, "queued", 0, 0, 0, 0, "not scheduled");
for (int i = 0; i < 101; i++) {
for (int i = 0; i < 11; i++) {
client.scheduleRequest();
requestComplete.await(10, TimeUnit.SECONDS);
tickerIncrement.set(new Duration(1, TimeUnit.SECONDS));
}
assertEquals(callback.getPages().size(), 0);
assertEquals(callback.getCompletedRequests(), 101);
assertEquals(callback.getCompletedRequests(), 11);
assertEquals(callback.getFinishedBuffers(), 0);
assertEquals(callback.getFailedBuffers(), 0);
assertEquals(callback.getFailedBuffers(), 2);
assertStatus(client, location, "queued", 0, 101, 101, 101, "not scheduled");
assertStatus(client, location, "queued", 0, 11, 11, 11, "not scheduled");
tickerIncrement.set(new Duration(301, TimeUnit.SECONDS));
client.scheduleRequest();
requestComplete.await(10, TimeUnit.SECONDS);
assertEquals(callback.getCompletedRequests(), 102);
assertEquals(callback.getCompletedRequests(), 12);
assertEquals(callback.getFinishedBuffers(), 0);
assertEquals(callback.getFailedBuffers(), 1);
assertEquals(callback.getFailedBuffers(), 3);
assertInstanceOf(callback.getFailure(), PageTransportTimeoutException.class);
assertContains(callback.getFailure().getMessage(), WORKER_NODE_ERROR + " (http://localhost:8080/0 - 102 failures,");
assertContains(callback.getFailure().getMessage(), WORKER_NODE_ERROR + " (http://localhost:8080/0 - 10 failures,");
}
@Test

5
presto-main/src/test/java/io/prestosql/operator/TestMergeOperator.java
View File

@ -34,9 +34,11 @@ import io.airlift.node.testing.TestingNodeModule;
import io.airlift.tracetoken.TraceTokenModule;
import io.prestosql.execution.Lifespan;
import io.prestosql.execution.QueryManagerConfig;
import io.prestosql.failuredetector.FailureDetectorManager;
import io.prestosql.failuredetector.FailureDetectorModule;
import io.prestosql.failuredetector.HeartbeatFailureDetector;
import io.prestosql.failuredetector.TestHeartbeatFailureDetector;
import io.prestosql.failuredetector.TimeoutFailureRetryFactory;
import io.prestosql.metadata.Split;
import io.prestosql.server.InternalCommunicationConfig;
import io.prestosql.spi.Page;
@ -139,7 +141,8 @@ public class TestMergeOperator
taskBuffers = CacheBuilder.newBuilder().build(CacheLoader.from(TestingTaskBuffer::new));
httpClient = new TestingHttpClient(new TestingExchangeHttpClientHandler(taskBuffers), executor);
exchangeClientFactory = new ExchangeClientFactory(new ExchangeClientConfig(), httpClient, executor, detector);
exchangeClientFactory = new ExchangeClientFactory(new ExchangeClientConfig(), httpClient, executor, new FailureDetectorManager(detector, "60s"));
FailureDetectorManager.addFailureRetryFactory(new TimeoutFailureRetryFactory());
orderingCompiler = new OrderingCompiler();
}

4
presto-main/src/test/java/io/prestosql/server/remotetask/TestBackoff.java
View File

@ -100,7 +100,7 @@ public class TestBackoff
TestingTicker ticker = new TestingTicker();
ticker.increment(1, NANOSECONDS);
Backoff backoff = new Backoff(3, new Duration(30, SECONDS), ticker, ImmutableList.of(new Duration(10, MILLISECONDS)));
MaxRetryBackoff backoff = new MaxRetryBackoff(new Duration(30, SECONDS), 10, ticker);
ticker.increment(10, MICROSECONDS);
// verify initial state
assertEquals(backoff.getFailureCount(), 0);
@ -120,7 +120,7 @@ public class TestBackoff
ticker.increment(1, SECONDS);
}
assertFalse(backoff.maxTried());
assertTrue(backoff.maxRetryDone());
assertFalse(backoff.timeout()); // 30 s should not elapse
ticker.increment(5, SECONDS);

6
presto-spi/src/main/java/io/prestosql/spi/Plugin.java
View File

@ -17,6 +17,7 @@ import io.prestosql.spi.block.BlockEncoding;
import io.prestosql.spi.connector.ConnectorFactory;
import io.prestosql.spi.cube.CubeProvider;
import io.prestosql.spi.eventlistener.EventListenerFactory;
import io.prestosql.spi.failuredetector.FailureRetryFactory;
import io.prestosql.spi.filesystem.HetuFileSystemClientFactory;
import io.prestosql.spi.function.FunctionNamespaceManagerFactory;
import io.prestosql.spi.heuristicindex.IndexFactory;
@ -127,6 +128,11 @@ public interface Plugin
return emptyList();
}
default Iterable<FailureRetryFactory> getFailureRetryFactory()
{
return emptyList();
}
default Iterable<HetuMetaStoreFactory> getHetuMetaStoreFactories()
{
return emptyList();

28
presto-spi/src/main/java/io/prestosql/spi/failuredetector/FailureRetryFactory.java Normal file
View File

@ -0,0 +1,28 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.spi.failuredetector;
import com.google.common.base.Ticker;
import java.util.Properties;
public interface FailureRetryFactory
{
FailureRetryPolicy getFailureRetryPolicy(Properties properties);
FailureRetryPolicy getFailureRetryPolicy(Properties properties, Ticker ticker);
String getName();
}

36
presto-spi/src/main/java/io/prestosql/spi/failuredetector/FailureRetryPolicy.java Normal file
View File

@ -0,0 +1,36 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.spi.failuredetector;
import io.prestosql.spi.HostAddress;
public interface FailureRetryPolicy
{
IBackoff getBackoff();
boolean hasFailed(HostAddress address);
String FD_RETRY_TYPE = "failure.recovery.retry.type";
String FD_RETRY_PROFILE = "failure.recovery.retry.profile";
String MAX_RETRY_COUNT = "max.retry.count";
String MAX_TIMEOUT_DURATION = "max.error.duration";
int DEFAULT_RETRY_COUNT = 100;
String DEFAULT_TIMEOUT_DURATION = "300s";
// FailureRetryPolicy type names, to be used in profile parameter failure.recovery.retry.type
String TIMEOUT = "timeout";
String MAXRETRY = "max-retry";
}

48
presto-spi/src/main/java/io/prestosql/spi/failuredetector/IBackoff.java Normal file
View File

@ -0,0 +1,48 @@
/*
* Licensed 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 io.prestosql.spi.failuredetector;
import com.google.common.collect.ImmutableList;
import io.airlift.units.Duration;
import java.util.List;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
public interface IBackoff
{
int MIN_RETRIES = 3;
List<Duration> DEFAULT_BACKOFF_DELAY_INTERVALS = ImmutableList.<Duration>builder()
.add(new Duration(0, MILLISECONDS))
.add(new Duration(50, MILLISECONDS))
.add(new Duration(100, MILLISECONDS))
.add(new Duration(200, MILLISECONDS))
.add(new Duration(500, MILLISECONDS))
.build();
boolean failure();
void startRequest();
long getBackoffDelayNanos();
void success();
long getFailureCount();
Duration getFailureDuration();
Duration getFailureRequestTimeTotal();
}