diff --git a/hetu-docs/en/admin/properties.md b/hetu-docs/en/admin/properties.md index 5075c5edc..5b925ebaa 100644 --- a/hetu-docs/en/admin/properties.md +++ b/hetu-docs/en/admin/properties.md @@ -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 `` set for this property has to correspond to `.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` diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/AbstractFailureRetryPolicy.java b/presto-main/src/main/java/io/prestosql/failuredetector/AbstractFailureRetryPolicy.java new file mode 100644 index 000000000..af1dbdabc --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/AbstractFailureRetryPolicy.java @@ -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); +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/FailureDetectorManager.java b/presto-main/src/main/java/io/prestosql/failuredetector/FailureDetectorManager.java new file mode 100644 index 000000000..fee6e8f8f --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/FailureDetectorManager.java @@ -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 failureDetectors = new ArrayList<>(); + + private static final Map failureRetryFactories = new ConcurrentHashMap<>(); + private static final Map 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 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 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); + } + } +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/FailureDetectorPlugin.java b/presto-main/src/main/java/io/prestosql/failuredetector/FailureDetectorPlugin.java new file mode 100644 index 000000000..ce6108c4d --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/FailureDetectorPlugin.java @@ -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 getFailureRetryFactory() + { + return ImmutableList.of(new TimeoutFailureRetryFactory(), new MaxRetryFailureRetryFactory()); + } +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/FailureRetryConfig.java b/presto-main/src/main/java/io/prestosql/failuredetector/FailureRetryConfig.java new file mode 100644 index 000000000..0005c1ac4 --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/FailureRetryConfig.java @@ -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; + } +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/FailureRetryPolicyFactory.java b/presto-main/src/main/java/io/prestosql/failuredetector/FailureRetryPolicyFactory.java new file mode 100644 index 000000000..96638600a --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/FailureRetryPolicyFactory.java @@ -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); +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/MaxRetryFailureRetryConfig.java b/presto-main/src/main/java/io/prestosql/failuredetector/MaxRetryFailureRetryConfig.java new file mode 100644 index 000000000..1d4da3e57 --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/MaxRetryFailureRetryConfig.java @@ -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; + } +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/MaxRetryFailureRetryFactory.java b/presto-main/src/main/java/io/prestosql/failuredetector/MaxRetryFailureRetryFactory.java new file mode 100644 index 000000000..0f4f3f82d --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/MaxRetryFailureRetryFactory.java @@ -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; + } +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/MaxRetryFailureRetryPolicy.java b/presto-main/src/main/java/io/prestosql/failuredetector/MaxRetryFailureRetryPolicy.java new file mode 100644 index 000000000..4c4e5dba9 --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/MaxRetryFailureRetryPolicy.java @@ -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(); + } +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/TimeoutFailureRetryConfig.java b/presto-main/src/main/java/io/prestosql/failuredetector/TimeoutFailureRetryConfig.java new file mode 100644 index 000000000..b24d0c29f --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/TimeoutFailureRetryConfig.java @@ -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); + } +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/TimeoutFailureRetryFactory.java b/presto-main/src/main/java/io/prestosql/failuredetector/TimeoutFailureRetryFactory.java new file mode 100644 index 000000000..87616e040 --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/TimeoutFailureRetryFactory.java @@ -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; + } +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/TimeoutFailureRetryPolicy.java b/presto-main/src/main/java/io/prestosql/failuredetector/TimeoutFailureRetryPolicy.java new file mode 100644 index 000000000..5db2246fd --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/TimeoutFailureRetryPolicy.java @@ -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(); + } +} diff --git a/presto-main/src/main/java/io/prestosql/operator/ExchangeClient.java b/presto-main/src/main/java/io/prestosql/operator/ExchangeClient.java index ee04c6d52..aaee6ae6a 100644 --- a/presto-main/src/main/java/io/prestosql/operator/ExchangeClient.java +++ b/presto-main/src/main/java/io/prestosql/operator/ExchangeClient.java @@ -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); diff --git a/presto-main/src/main/java/io/prestosql/operator/ExchangeClientConfig.java b/presto-main/src/main/java/io/prestosql/operator/ExchangeClientConfig.java index a669107f4..a1a19b6ab 100644 --- a/presto-main/src/main/java/io/prestosql/operator/ExchangeClientConfig.java +++ b/presto-main/src/main/java/io/prestosql/operator/ExchangeClientConfig.java @@ -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() diff --git a/presto-main/src/main/java/io/prestosql/operator/ExchangeClientFactory.java b/presto-main/src/main/java/io/prestosql/operator/ExchangeClientFactory.java index ce599ce46..25c556c6c 100644 --- a/presto-main/src/main/java/io/prestosql/operator/ExchangeClientFactory.java +++ b/presto-main/src/main/java/io/prestosql/operator/ExchangeClientFactory.java @@ -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); } } diff --git a/presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java b/presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java index c77404e5d..b9e6092c7 100644 --- a/presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java +++ b/presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java @@ -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); } diff --git a/presto-main/src/main/java/io/prestosql/server/PluginManager.java b/presto-main/src/main/java/io/prestosql/server/PluginManager.java index ea281a49d..3f9ff137e 100644 --- a/presto-main/src/main/java/io/prestosql/server/PluginManager.java +++ b/presto-main/src/main/java/io/prestosql/server/PluginManager.java @@ -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); } diff --git a/presto-main/src/main/java/io/prestosql/server/PrestoServer.java b/presto-main/src/main/java/io/prestosql/server/PrestoServer.java index 9ed92f7b6..7880e9415 100755 --- a/presto-main/src/main/java/io/prestosql/server/PrestoServer.java +++ b/presto-main/src/main/java/io/prestosql/server/PrestoServer.java @@ -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( diff --git a/presto-main/src/main/java/io/prestosql/server/ServerMainModule.java b/presto-main/src/main/java/io/prestosql/server/ServerMainModule.java index cdbdb8120..c5769d78b 100644 --- a/presto-main/src/main/java/io/prestosql/server/ServerMainModule.java +++ b/presto-main/src/main/java/io/prestosql/server/ServerMainModule.java @@ -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) diff --git a/presto-main/src/main/java/io/prestosql/server/remotetask/Backoff.java b/presto-main/src/main/java/io/prestosql/server/remotetask/Backoff.java index 86d8605e8..6b2492fa6 100644 --- a/presto-main/src/main/java/io/prestosql/server/remotetask/Backoff.java +++ b/presto-main/src/main/java/io/prestosql/server/remotetask/Backoff.java @@ -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 DEFAULT_BACKOFF_DELAY_INTERVALS = ImmutableList.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 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; diff --git a/presto-main/src/main/java/io/prestosql/server/remotetask/MaxRetryBackoff.java b/presto-main/src/main/java/io/prestosql/server/remotetask/MaxRetryBackoff.java new file mode 100644 index 000000000..ac98f9df5 --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/server/remotetask/MaxRetryBackoff.java @@ -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; + } +} diff --git a/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java b/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java index c46315ac8..def85d797 100644 --- a/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java +++ b/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java @@ -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()); diff --git a/presto-main/src/test/java/io/prestosql/failuredetector/TestFailureDetectionManager.java b/presto-main/src/test/java/io/prestosql/failuredetector/TestFailureDetectionManager.java new file mode 100644 index 000000000..44df238ed --- /dev/null +++ b/presto-main/src/test/java/io/prestosql/failuredetector/TestFailureDetectionManager.java @@ -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 frConfigs = FailureDetectorManager.getAvailableFrConfigs(); + assertNotNull(frConfigs.get("default1")); + assertNotNull(frConfigs.get("test2")); + assertNotNull(frConfigs.get("default")); + } + + @Test + public void testDefaultRetryProfile() + { + Map 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 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 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)); + } +} diff --git a/presto-main/src/test/java/io/prestosql/operator/TestExchangeClient.java b/presto-main/src/test/java/io/prestosql/operator/TestExchangeClient.java index 3bd7ecf24..5bfd4888c 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestExchangeClient.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestExchangeClient.java @@ -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(); diff --git a/presto-main/src/test/java/io/prestosql/operator/TestExchangeClientConfig.java b/presto-main/src/test/java/io/prestosql/operator/TestExchangeClientConfig.java index 9d3492a68..bdbb8cb07 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestExchangeClientConfig.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestExchangeClientConfig.java @@ -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 properties = new ImmutableMap.Builder() .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); } diff --git a/presto-main/src/test/java/io/prestosql/operator/TestExchangeOperator.java b/presto-main/src/test/java/io/prestosql/operator/TestExchangeOperator.java index 08d04710e..7ebebd819 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestExchangeOperator.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestExchangeOperator.java @@ -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) diff --git a/presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java b/presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java index 549d7dcf7..6e77fd0b1 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java @@ -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 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 factories = FailureDetectorManager.getFailureRetryFactories(); + for (Map.Entry e : factories.entrySet()) { + log.info(e.getKey()); + } + } + + private void printGfConfigs() + { + log.info("the following profiles are present in the factory"); + Map frConfigs = FailureDetectorManager.getAvailableFrConfigs(); + for (Map.Entry 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 diff --git a/presto-main/src/test/java/io/prestosql/operator/TestMergeOperator.java b/presto-main/src/test/java/io/prestosql/operator/TestMergeOperator.java index a7aaa3c10..f3a7cf58c 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestMergeOperator.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestMergeOperator.java @@ -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(); } diff --git a/presto-main/src/test/java/io/prestosql/server/remotetask/TestBackoff.java b/presto-main/src/test/java/io/prestosql/server/remotetask/TestBackoff.java index 267cd9b13..9a46ec56b 100644 --- a/presto-main/src/test/java/io/prestosql/server/remotetask/TestBackoff.java +++ b/presto-main/src/test/java/io/prestosql/server/remotetask/TestBackoff.java @@ -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); diff --git a/presto-spi/src/main/java/io/prestosql/spi/Plugin.java b/presto-spi/src/main/java/io/prestosql/spi/Plugin.java index 0c2fa2d09..d9d8c69f3 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/Plugin.java +++ b/presto-spi/src/main/java/io/prestosql/spi/Plugin.java @@ -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 getFailureRetryFactory() + { + return emptyList(); + } + default Iterable getHetuMetaStoreFactories() { return emptyList(); diff --git a/presto-spi/src/main/java/io/prestosql/spi/failuredetector/FailureRetryFactory.java b/presto-spi/src/main/java/io/prestosql/spi/failuredetector/FailureRetryFactory.java new file mode 100644 index 000000000..dcfe2bdcd --- /dev/null +++ b/presto-spi/src/main/java/io/prestosql/spi/failuredetector/FailureRetryFactory.java @@ -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(); +} diff --git a/presto-spi/src/main/java/io/prestosql/spi/failuredetector/FailureRetryPolicy.java b/presto-spi/src/main/java/io/prestosql/spi/failuredetector/FailureRetryPolicy.java new file mode 100644 index 000000000..4148eda6c --- /dev/null +++ b/presto-spi/src/main/java/io/prestosql/spi/failuredetector/FailureRetryPolicy.java @@ -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"; +} diff --git a/presto-spi/src/main/java/io/prestosql/spi/failuredetector/IBackoff.java b/presto-spi/src/main/java/io/prestosql/spi/failuredetector/IBackoff.java new file mode 100644 index 000000000..9f427ea5d --- /dev/null +++ b/presto-spi/src/main/java/io/prestosql/spi/failuredetector/IBackoff.java @@ -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 DEFAULT_BACKOFF_DELAY_INTERVALS = ImmutableList.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(); +}