jihoonson closed pull request #6033: Synchronize scheduled poll() calls in SQLMetadataRuleManager to prevent flakiness in SqlMetadataRuleManagerTest URL: https://github.com/apache/incubator-druid/pull/6033
This is a PR merged from a forked repository. As GitHub hides the original diff on merge, it is displayed below for the sake of provenance: As this is a foreign pull request (from a fork), the diff is supplied below (as it won't show otherwise due to GitHub magic): diff --git a/server/src/main/java/io/druid/metadata/SQLMetadataRuleManager.java b/server/src/main/java/io/druid/metadata/SQLMetadataRuleManager.java index 4883a8e1100..6900dff018c 100644 --- a/server/src/main/java/io/druid/metadata/SQLMetadataRuleManager.java +++ b/server/src/main/java/io/druid/metadata/SQLMetadataRuleManager.java @@ -27,9 +27,6 @@ import com.google.common.collect.ImmutableMap; import com.google.common.collect.Lists; import com.google.common.collect.Maps; -import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.ListeningScheduledExecutorService; -import com.google.common.util.concurrent.MoreExecutors; import com.google.inject.Inject; import io.druid.audit.AuditEntry; import io.druid.audit.AuditInfo; @@ -63,6 +60,7 @@ import java.util.Arrays; import java.util.List; import java.util.Map; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -72,7 +70,6 @@ public class SQLMetadataRuleManager implements MetadataRuleManager { - public static void createDefaultRule( final IDBI dbi, final String ruleTable, @@ -142,13 +139,19 @@ public Void withHandle(Handle handle) throws Exception private final AuditManager auditManager; private final Object lock = new Object(); - - private volatile boolean started = false; - - private volatile ListeningScheduledExecutorService exec = null; - private volatile ListenableFuture<?> future = null; - - private volatile long retryStartTime = 0; + /** The number of times this SQLMetadataRuleManager was started. */ + private long startCount = 0; + /** + * Equal to the current {@link #startCount} value, if the SQLMetadataRuleManager is currently started; -1 if + * currently stopped. + * + * This field is used to implement a simple stamp mechanism instead of just a boolean "started" flag to prevent + * the theoretical situation of two tasks scheduled in {@link #start()} calling {@link #poll()} concurrently, if + * the sequence of {@link #start()} - {@link #stop()} - {@link #start()} actions occurs quickly. + */ + private long currentStartOrder = -1; + private ScheduledExecutorService exec = null; + private long retryStartTime = 0; @Inject public SQLMetadataRuleManager( @@ -169,9 +172,7 @@ public SQLMetadataRuleManager( Preconditions.checkNotNull(config.getAlertThreshold().toStandardDuration()); Preconditions.checkNotNull(config.getPollDuration().toStandardDuration()); - this.rules = new AtomicReference<>( - ImmutableMap.<String, List<Rule>>of() - ); + this.rules = new AtomicReference<>(ImmutableMap.of()); } @Override @@ -179,21 +180,34 @@ public SQLMetadataRuleManager( public void start() { synchronized (lock) { - if (started) { + if (currentStartOrder >= 0) { return; } - exec = MoreExecutors.listeningDecorator(Execs.scheduledSingleThreaded("DatabaseRuleManager-Exec--%d")); + startCount++; + currentStartOrder = startCount; + long localStartedOrder = currentStartOrder; + + exec = Execs.scheduledSingleThreaded("DatabaseRuleManager-Exec--%d"); createDefaultRule(dbi, getRulesTable(), config.getDefaultRule(), jsonMapper); - future = exec.scheduleWithFixedDelay( + exec.scheduleWithFixedDelay( new Runnable() { @Override public void run() { try { - poll(); + // poll() is synchronized together with start() and stop() to ensure that when stop() exists, poll() + // won't actually run anymore after that (it could only enter the syncrhonized section and exit + // immediately because the localStartedOrder doesn't match the new currentStartOrder). It's needed + // to avoid flakiness in SQLMetadataRuleManagerTest. + // See https://github.com/apache/incubator-druid/issues/6028 + synchronized (lock) { + if (localStartedOrder == currentStartOrder) { + poll(); + } + } } catch (Exception e) { log.error(e, "uncaught exception in rule manager polling thread"); @@ -204,8 +218,6 @@ public void run() config.getPollDuration().toStandardDuration().getMillis(), TimeUnit.MILLISECONDS ); - - started = true; } } @@ -214,15 +226,12 @@ public void run() public void stop() { synchronized (lock) { - if (!started) { + if (currentStartOrder == -1) { return; } - - rules.set(ImmutableMap.<String, List<Rule>>of()); - - future.cancel(false); - future = null; - started = false; + rules.set(ImmutableMap.of()); + currentStartOrder = -1; + // This call cancels the periodic poll() task, scheduled in start(). exec.shutdownNow(); exec = null; } ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services --------------------------------------------------------------------- To unsubscribe, e-mail: dev-unsubscr...@druid.apache.org For additional commands, e-mail: dev-h...@druid.apache.org