Skip to content

[FLINK-40143][core] Fix displayed restart strategy for config option based restart strategies #28736

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Open
csurong wants to merge 1 commit into
base: release-1.20
Choose a base branch
from
+110 −5
Open

[FLINK-40143][core] Fix displayed restart strategy for config option based restart strategies #28736

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 16 additions & 5 deletions flink-core/src/main/java/org/apache/flink/api/common/ArchivedExecutionConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.apache.flink.api.common;

import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;

import java.io.Serializable;
import java.util.Collections;
Expand All @@ -44,11 +45,7 @@ public class ArchivedExecutionConfig implements Serializable {

public ArchivedExecutionConfig(ExecutionConfig ec) {
executionMode = ec.getExecutionMode().name();
if (ec.getRestartStrategy() != null) {
restartStrategyDescription = ec.getRestartStrategy().getDescription();
} else {
restartStrategyDescription = "default";
}
restartStrategyDescription = getRestartStrategyDescription(ec);
maxParallelism = ec.getMaxParallelism();
parallelism = ec.getParallelism();
objectReuseEnabled = ec.isObjectReuseEnabled();
Expand All @@ -60,6 +57,20 @@ public ArchivedExecutionConfig(ExecutionConfig ec) {
}
}

private static String getRestartStrategyDescription(ExecutionConfig ec) {
final RestartStrategies.RestartStrategyConfiguration restartStrategy =
ec.getRestartStrategy();
if (restartStrategy != null
&& !(restartStrategy
instanceof RestartStrategies.FallbackRestartStrategyConfiguration)) {
return restartStrategy.getDescription();
}

return RestartStrategies.fromConfiguration(ec.toConfiguration())
.map(RestartStrategies.RestartStrategyConfiguration::getDescription)
.orElse(restartStrategy != null ? restartStrategy.getDescription() : "default");
}

public ArchivedExecutionConfig(
String executionMode,
String restartStrategyDescription,
Expand Down
94 changes: 94 additions & 0 deletions flink-core/src/test/java/org/apache/flink/api/common/ExecutionConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.RestartStrategyOptions;
import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.util.SerializedValue;

Expand All @@ -37,6 +38,7 @@
import org.junit.jupiter.api.Test;

import java.io.Serializable;
import java.time.Duration;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
Expand Down Expand Up @@ -175,6 +177,98 @@ void testReadingDefaultConfig() {
assertThat(executionConfig).isEqualTo(new ExecutionConfig());
}

@Test
void testArchiveUsesRestartStrategyConfiguredViaConfigOptions() {
assertThat(archiveRestartStrategyDescription(fixedDelayRestartConfiguration()))
.isEqualTo("Restart with fixed delay (PT1M). #10 restart attempts.");

Configuration noRestartConfiguration = new Configuration();
noRestartConfiguration.set(
RestartStrategyOptions.RESTART_STRATEGY,
RestartStrategyOptions.RestartStrategyType.NO_RESTART_STRATEGY.getMainValue());
assertThat(archiveRestartStrategyDescription(noRestartConfiguration))
.isEqualTo("Restart deactivated.");

Configuration failureRateConfiguration = new Configuration();
failureRateConfiguration.set(
RestartStrategyOptions.RESTART_STRATEGY,
RestartStrategyOptions.RestartStrategyType.FAILURE_RATE.getMainValue());
failureRateConfiguration.set(
RestartStrategyOptions.RESTART_STRATEGY_FAILURE_RATE_MAX_FAILURES_PER_INTERVAL, 7);
failureRateConfiguration.set(
RestartStrategyOptions.RESTART_STRATEGY_FAILURE_RATE_FAILURE_RATE_INTERVAL,
Duration.ofMinutes(2));
failureRateConfiguration.set(
RestartStrategyOptions.RESTART_STRATEGY_FAILURE_RATE_DELAY, Duration.ofSeconds(3));
assertThat(archiveRestartStrategyDescription(failureRateConfiguration))
.isEqualTo(
"Failure rate restart with maximum of 7 failures within interval PT2M and fixed delay PT3S.");

Configuration exponentialDelayConfiguration = new Configuration();
exponentialDelayConfiguration.set(
RestartStrategyOptions.RESTART_STRATEGY,
RestartStrategyOptions.RestartStrategyType.EXPONENTIAL_DELAY.getMainValue());
exponentialDelayConfiguration.set(
RestartStrategyOptions.RESTART_STRATEGY_EXPONENTIAL_DELAY_INITIAL_BACKOFF,
Duration.ofSeconds(1));
exponentialDelayConfiguration.set(
RestartStrategyOptions.RESTART_STRATEGY_EXPONENTIAL_DELAY_MAX_BACKOFF,
Duration.ofMinutes(1));
exponentialDelayConfiguration.set(
RestartStrategyOptions.RESTART_STRATEGY_EXPONENTIAL_DELAY_BACKOFF_MULTIPLIER, 2.0);
exponentialDelayConfiguration.set(
RestartStrategyOptions.RESTART_STRATEGY_EXPONENTIAL_DELAY_RESET_BACKOFF_THRESHOLD,
Duration.ofHours(1));
exponentialDelayConfiguration.set(
RestartStrategyOptions.RESTART_STRATEGY_EXPONENTIAL_DELAY_JITTER_FACTOR, 0.3);
assertThat(archiveRestartStrategyDescription(exponentialDelayConfiguration))
.isEqualTo(
"Restart with exponential delay: starting at PT1S, increasing by 2.000000, with maximum PT1M. "
+ "Delay resets after PT1H with jitter 0.300000");
}

@Test
void testArchivePrefersExplicitRestartStrategyOverConfigOptions() {
ExecutionConfig executionConfig = new ExecutionConfig(fixedDelayRestartConfiguration());
executionConfig.setRestartStrategy(RestartStrategies.noRestart());

assertThat(executionConfig.archive().getRestartStrategyDescription())
.isEqualTo("Restart deactivated.");
}

@Test
void testArchiveUsesFallbackRestartStrategyWithoutConfigOptions() {
assertThat(new ExecutionConfig().archive().getRestartStrategyDescription())
.isEqualTo("Cluster level default restart strategy");
}

@Test
void testArchiveKeepsLegacyExecutionRetryRestartStrategy() {
ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.setNumberOfExecutionRetries(4);
executionConfig.setExecutionRetryDelay(123);

assertThat(executionConfig.archive().getRestartStrategyDescription())
.isEqualTo("Restart with fixed delay (PT0.123S). #4 restart attempts.");
}

private String archiveRestartStrategyDescription(Configuration configuration) {
ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.configure(configuration, ExecutionConfigTest.class.getClassLoader());
return executionConfig.archive().getRestartStrategyDescription();
}

private Configuration fixedDelayRestartConfiguration() {
Configuration configuration = new Configuration();
configuration.set(
RestartStrategyOptions.RESTART_STRATEGY,
RestartStrategyOptions.RestartStrategyType.FIXED_DELAY.getMainValue());
configuration.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 10);
configuration.set(
RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY, Duration.ofMinutes(1));
return configuration;
}

@Test
void testLoadingRegisteredKryoTypesFromConfiguration() {
ExecutionConfig configFromSetters = new ExecutionConfig();
Expand Down