diff --git a/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/event/retry/BackoffRetrySupport.java b/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/event/retry/BackoffRetrySupport.java new file mode 100644 index 0000000000..0046654f86 --- /dev/null +++ b/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/event/retry/BackoffRetrySupport.java @@ -0,0 +1,101 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.ranger.resource.mapper.event.retry; + +import java.util.concurrent.ThreadLocalRandom; +import lombok.extern.slf4j.Slf4j; +import org.apache.ranger.resource.mapper.utils.ThrowingRunnable; + +/** + * Retry helper that never gives up. It runs the action and, if it fails, waits and tries + * again until the action works or the thread is interrupted. The wait grows after each + * failure (exponential backoff), but it is capped at a maximum, and a small random jitter + * is added so many clients do not retry at the same moment. + */ +@Slf4j +public class BackoffRetrySupport implements RetrySupport { + + private final long baseIntervalMs; + private final long maxIntervalMs; + private final RetrySleeper retrySleeper; + + /** + * The config values are already checked in the config class. Here we only keep a safe floor + * (base at least 1 ms, max not smaller than base), so this helper can never break on a wrong + * wait time, even if it is created with bad values from some other place. + * + * @param baseIntervalMs wait time for the first retry, in milliseconds + * @param maxIntervalMs the biggest possible wait time, in milliseconds + * @param retrySleeper how to wait between the tries (usually {@code Thread::sleep}) + */ + public BackoffRetrySupport(long baseIntervalMs, long maxIntervalMs, RetrySleeper retrySleeper) { + this.baseIntervalMs = Math.max(1, baseIntervalMs); + this.maxIntervalMs = Math.max(this.baseIntervalMs, maxIntervalMs); + this.retrySleeper = retrySleeper; + } + + /** + * Runs the action. If the action throws, it waits (see {@link #backoffWithJitterMs(int)}) + * and runs the action one more time. It repeats this forever until the action does not throw. + * + * @param action the work to run; it can throw any exception + * @throws RetryException only when the thread is interrupted during the wait. Before it throws, + * it sets the interrupt flag again, so the caller can also see the interrupt. + */ + @Override + public void withRetries(ThrowingRunnable action) throws RetryException { + int attempt = 0; + while (true) { + try { + action.run(); + return; + } catch (Exception e) { + long delayMs = backoffWithJitterMs(attempt++); + log.warn("Attempt failed, retrying in {} ms", delayMs, e); + try { + retrySleeper.sleepForMs(delayMs); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw new RetryException("Interrupted while backing off before retry", interrupted); + } + } + } + } + + /** + * Calculates how long to wait before the next try. + * + *

The base value is doubled one time for every attempt (base, base*2, base*4 ...), but the + * result is never bigger than the maximum. After that a random jitter is added: the returned + * value is between the half of the capped value and the full capped value.

+ * + * @param attempt how many times we already failed (the first call uses 0) + * @return the wait time in milliseconds + */ + private long backoffWithJitterMs(int attempt) { + long delay = baseIntervalMs; + for (int i = 0; i < attempt && delay < maxIntervalMs; i++) { + delay <<= 1; + } + long capped = Math.min(delay, maxIntervalMs); + long half = capped / 2; + return half + ThreadLocalRandom.current().nextLong(half + 1); + } +} diff --git a/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/HiveResourceMappingManager.java b/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/HiveResourceMappingManager.java index 7e74d11752..334aa55d4a 100644 --- a/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/HiveResourceMappingManager.java +++ b/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/HiveResourceMappingManager.java @@ -36,6 +36,7 @@ import org.apache.ranger.resource.mapper.event.ResourceDiffCollector; import org.apache.ranger.resource.mapper.event.ResourceDiffHandler; import org.apache.ranger.resource.mapper.event.ResourceDiffSource; +import org.apache.ranger.resource.mapper.event.retry.BackoffRetrySupport; import org.apache.ranger.resource.mapper.event.retry.PolicyBasedRetrySupport; import org.apache.ranger.resource.mapper.event.retry.ResourceMapperRetryPolicy; import org.apache.ranger.resource.mapper.event.retry.RetryPolicyFactory; @@ -104,8 +105,9 @@ private ResourceDiffSource buildEventFetcher(RetryPolicyFactory retryPolicyFacto HiveMetastoreSnapshotFetcher snapshotFetcher = buildSnapshotEventFetcher(hiveMetaStoreClient, retrySupport, hiveAuthenticator, config); + RetrySupport eventFetcherRetrySupport = buildHiveListenerBackoffRetrySupport(config); HiveMetastoreEventFetcher eventFetcher = buildHiveMetastoreEventFetcher( - hiveMetaStoreClient, retrySupport, hiveAuthenticator, config); + hiveMetaStoreClient, eventFetcherRetrySupport, hiveAuthenticator, config); return CompositeHiveMetastoreFetcher.builder() .metaStoreClient(hiveMetaStoreClient) @@ -163,6 +165,13 @@ private RetrySupport buildHiveListenerRetrySupport( return new PolicyBasedRetrySupport(retryPolicy, Thread::sleep); } + private RetrySupport buildHiveListenerBackoffRetrySupport(HiveResourceMappingManagerConfig config) { + return new BackoffRetrySupport( + config.getHmsReconnectBaseIntervalMs(), + config.getHmsReconnectMaxIntervalMs(), + Thread::sleep); + } + private RetrySupport buildEventApplierRetrySupport( RetryPolicyFactory retryPolicyFactory, ResourceMappingManagerConfig config) { diff --git a/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/config/ConfigurationKeys.java b/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/config/ConfigurationKeys.java index 679e8c90ab..52461c1de1 100644 --- a/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/config/ConfigurationKeys.java +++ b/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/config/ConfigurationKeys.java @@ -37,6 +37,12 @@ public class ConfigurationKeys { public static final String HMS_MAX_RETRIES = "ranger.rmm.hms.retry.max"; public static final int HMS_MAX_RETRIES_DEFAULT = 10; + public static final String HMS_RECONNECT_BASE_INTERVAL_MS = "ranger.rmm.hms.reconnect.base.interval.ms"; + public static final long HMS_RECONNECT_BASE_INTERVAL_MS_DEFAULT = 1000L; + + public static final String HMS_RECONNECT_MAX_INTERVAL_MS = "ranger.rmm.hms.reconnect.max.interval.ms"; + public static final long HMS_RECONNECT_MAX_INTERVAL_MS_DEFAULT = 300000L; + public static final String HMS_FULL_SYNC = "ranger.rmm.hms.sync.full"; public static final boolean HMS_FULL_SYNC_DEFAULT = false; } diff --git a/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/config/HiveResourceMappingManagerConfig.java b/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/config/HiveResourceMappingManagerConfig.java index fddbd25d09..2168631714 100644 --- a/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/config/HiveResourceMappingManagerConfig.java +++ b/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/config/HiveResourceMappingManagerConfig.java @@ -27,14 +27,20 @@ import static org.apache.ranger.resource.mapper.hive.config.ConfigurationKeys.HMS_FULL_SYNC_DEFAULT; import static org.apache.ranger.resource.mapper.hive.config.ConfigurationKeys.HMS_MAX_RETRIES; import static org.apache.ranger.resource.mapper.hive.config.ConfigurationKeys.HMS_MAX_RETRIES_DEFAULT; +import static org.apache.ranger.resource.mapper.hive.config.ConfigurationKeys.HMS_RECONNECT_BASE_INTERVAL_MS; +import static org.apache.ranger.resource.mapper.hive.config.ConfigurationKeys.HMS_RECONNECT_BASE_INTERVAL_MS_DEFAULT; +import static org.apache.ranger.resource.mapper.hive.config.ConfigurationKeys.HMS_RECONNECT_MAX_INTERVAL_MS; +import static org.apache.ranger.resource.mapper.hive.config.ConfigurationKeys.HMS_RECONNECT_MAX_INTERVAL_MS_DEFAULT; import static org.apache.ranger.resource.mapper.hive.config.ConfigurationKeys.HMS_RETRY_INTERVAL_MS; import static org.apache.ranger.resource.mapper.hive.config.ConfigurationKeys.HMS_RETRY_INTERVAL_MS_DEFAULT; import static org.apache.ranger.resource.mapper.hive.config.ConfigurationKeys.HMS_RETRY_STRATEGY; import static org.apache.ranger.resource.mapper.hive.config.ConfigurationKeys.HMS_RETRY_STRATEGY_DEFAULT; +import lombok.extern.slf4j.Slf4j; import org.apache.ranger.resource.mapper.config.ResourceMappingManagerConfig; import org.apache.ranger.resource.mapper.event.retry.RetryStrategy; +@Slf4j public class HiveResourceMappingManagerConfig extends ResourceMappingManagerConfig { private static final String HIVE_CONFIG_FILE = "hive-site.xml"; @@ -58,6 +64,18 @@ public long getHiveListenerRetryIntervalMs() { return getLong(HMS_RETRY_INTERVAL_MS, HMS_RETRY_INTERVAL_MS_DEFAULT); } + public long getHmsReconnectBaseIntervalMs() { + return validate( + getLong(HMS_RECONNECT_BASE_INTERVAL_MS, HMS_RECONNECT_BASE_INTERVAL_MS_DEFAULT), + HMS_RECONNECT_BASE_INTERVAL_MS_DEFAULT); + } + + public long getHmsReconnectMaxIntervalMs() { + return validate( + getLong(HMS_RECONNECT_MAX_INTERVAL_MS, HMS_RECONNECT_MAX_INTERVAL_MS_DEFAULT), + HMS_RECONNECT_MAX_INTERVAL_MS_DEFAULT); + } + public int getHiveListenerMaxRetries() { return getInt(HMS_MAX_RETRIES, HMS_MAX_RETRIES_DEFAULT); } @@ -65,4 +83,12 @@ public int getHiveListenerMaxRetries() { public boolean isFullMetastoreSync() { return getBoolean(HMS_FULL_SYNC, HMS_FULL_SYNC_DEFAULT); } + + private static long validate(long value, long defaultValue) { + if (value < 1) { + log.warn("Config value {} ms is not valid, using default {} ms instead", value, defaultValue); + return defaultValue; + } + return value; + } } diff --git a/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/event/fetch/HiveMetastoreEventFetcher.java b/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/event/fetch/HiveMetastoreEventFetcher.java index e2a640bea4..272e8734e0 100644 --- a/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/event/fetch/HiveMetastoreEventFetcher.java +++ b/resource-mapping-manager/src/main/java/org/apache/ranger/resource/mapper/hive/event/fetch/HiveMetastoreEventFetcher.java @@ -80,6 +80,8 @@ public class HiveMetastoreEventFetcher extends BaseHiveMetastoreFetcher { @Getter(PROTECTED) private final Long endEventId; + private boolean metaStoreClientConnected = true; + @lombok.Builder( builderClassName = "Builder", toBuilder = true @@ -117,7 +119,7 @@ public BlockingQueue pollAsync(long fromEventId) throw if (pollStarted.compareAndSet(false, true)) { authenticator.login(); lastHandledEventId = fromEventId; - executor.scheduleAtFixedRate(this::pollRecordsBatch, + executor.scheduleWithFixedDelay(this::pollRecordsBatch, 0L, fetchPeriodMs, TimeUnit.MILLISECONDS); } return outputQueue; @@ -138,13 +140,18 @@ void pollRecordsBatch() { close(); } } catch (Exception exception) { - log.error("Exiting HiveMetastoreEventFetcher due to error", exception); - close(); + log.error("Stopped polling HMS events", exception); } } private void pollRecordsBatchAction() { try { + if (!metaStoreClientConnected) { + metaStoreClient.reconnect(); + metaStoreClientConnected = true; + log.info("Reconnected to Hive Metastore"); + } + List events = metaStoreClient.getNextNotification( lastHandledEventId, eventBatchSize, @@ -166,6 +173,7 @@ private void pollRecordsBatchAction() { handle(event); } } catch (Exception e) { + metaStoreClientConnected = false; throw new RuntimeException("Error polling records batch from Hive Metastore", e); } } @@ -348,7 +356,7 @@ public HiveMetastoreEventFetcher toFiniteFetcher(long endEventId) { @Override public void close() { if (executor != null) { - executor.shutdown(); + executor.shutdownNow(); } try { diff --git a/resource-mapping-manager/src/test/java/org/apache/ranger/resource/mapper/event/retry/BackoffRetrySupportTest.java b/resource-mapping-manager/src/test/java/org/apache/ranger/resource/mapper/event/retry/BackoffRetrySupportTest.java new file mode 100644 index 0000000000..a4b15e5062 --- /dev/null +++ b/resource-mapping-manager/src/test/java/org/apache/ranger/resource/mapper/event/retry/BackoffRetrySupportTest.java @@ -0,0 +1,158 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.ranger.resource.mapper.event.retry; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.ArrayList; +import java.util.List; +import lombok.RequiredArgsConstructor; +import org.apache.ranger.resource.mapper.utils.ThrowingRunnable; +import org.junit.jupiter.api.Test; + +class BackoffRetrySupportTest { + + private static final long BASE_MS = 100L; + private static final long MAX_MS = 1000L; + + // When the action works from the first time, we run it one time and we do not wait. + @Test + public void testRunsOnceWhenActionSucceeds() throws RetryException { + RecordingSleeper sleeper = new RecordingSleeper(); + BackoffRetrySupport retries = new BackoffRetrySupport(BASE_MS, MAX_MS, sleeper); + + FailingAction action = new FailingAction(0); + retries.withRetries(action); + + assertEquals(1, action.attempts); + assertTrue(sleeper.delays.isEmpty()); + } + + // When the action fails some times and then works, we retry until it works. + @Test + public void testRetriesUntilActionSucceeds() throws RetryException { + RecordingSleeper sleeper = new RecordingSleeper(); + BackoffRetrySupport retries = new BackoffRetrySupport(BASE_MS, MAX_MS, sleeper); + + FailingAction action = new FailingAction(5); + retries.withRetries(action); + + assertEquals(6, action.attempts); + assertEquals(5, sleeper.delays.size()); + } + + // The helper never gives up: even after many failures it still retries (a fixed-limit + // retry would already throw here). This is the point of the fix for ADPS-1428. + @Test + public void testNeverGivesUp() throws RetryException { + RecordingSleeper sleeper = new RecordingSleeper(); + BackoffRetrySupport retries = new BackoffRetrySupport(BASE_MS, MAX_MS, sleeper); + + FailingAction action = new FailingAction(100); + retries.withRetries(action); + + assertEquals(101, action.attempts); + assertEquals(100, sleeper.delays.size()); + } + + // The wait time grows after each failure and never becomes bigger than the maximum. + // With jitter every wait must be between half of the cap and the full cap. + @Test + public void testDelayGrowsAndIsCapped() throws RetryException { + RecordingSleeper sleeper = new RecordingSleeper(); + BackoffRetrySupport retries = new BackoffRetrySupport(BASE_MS, MAX_MS, sleeper); + + retries.withRetries(new FailingAction(6)); + + assertEquals(6, sleeper.delays.size()); + for (int attempt = 0; attempt < sleeper.delays.size(); attempt++) { + long cap = expectedCap(attempt); + long delay = sleeper.delays.get(attempt); + assertTrue(delay >= cap / 2, "delay " + delay + " is below half of cap " + cap); + assertTrue(delay <= cap, "delay " + delay + " is above cap " + cap); + } + // last waits must already sit at the maximum cap + assertEquals(MAX_MS, expectedCap(4)); + assertEquals(MAX_MS, expectedCap(5)); + } + + // If the thread is interrupted while we wait, we stop, throw RetryException and keep + // the interrupt flag set, so the caller (for example on shutdown) can also see it. + @Test + public void testInterruptDuringWaitStopsRetrying() { + RetrySleeper interruptingSleeper = durationMs -> { + throw new InterruptedException(); + }; + BackoffRetrySupport retries = new BackoffRetrySupport(BASE_MS, MAX_MS, interruptingSleeper); + + assertThrows(RetryException.class, () -> retries.withRetries(new FailingAction(1))); + assertTrue(Thread.interrupted(), "the interrupt flag must be set again before throwing"); + } + + // Bad config values (negative, zero, or max below base) must not crash the helper. + // They are fixed to safe values in the constructor, so every wait time is still valid. + @Test + public void testInvalidConfigIsFixedAndDoesNotCrash() throws RetryException { + RecordingSleeper sleeper = new RecordingSleeper(); + BackoffRetrySupport retries = new BackoffRetrySupport(-5L, -1L, sleeper); + + retries.withRetries(new FailingAction(3)); + + assertEquals(3, sleeper.delays.size()); + for (long delay : sleeper.delays) { + assertTrue(delay >= 0, "wait time must not be negative: " + delay); + } + } + + // Same growth rule as in BackoffRetrySupport: double the base for each attempt, but not + // bigger than the maximum. + private long expectedCap(int attempt) { + long delay = BASE_MS; + for (int i = 0; i < attempt && delay < MAX_MS; i++) { + delay <<= 1; + } + return Math.min(delay, MAX_MS); + } + + private static class RecordingSleeper implements RetrySleeper { + private final List delays = new ArrayList<>(); + + @Override + public void sleepForMs(long durationMs) { + delays.add(durationMs); + } + } + + @RequiredArgsConstructor + private static class FailingAction implements ThrowingRunnable { + + private final int failsBeforeRun; + private int attempts; + + @Override + public void run() throws Exception { + if (++attempts <= failsBeforeRun) { + throw new Exception(); + } + } + } +}