diff --git a/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/cli/GlueSyncCli.java b/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/cli/GlueSyncCli.java index 9423fcce..6eb6fd1e 100644 --- a/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/cli/GlueSyncCli.java +++ b/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/cli/GlueSyncCli.java @@ -1,5 +1,5 @@ /** - * Copyright (C) 2018-2025 Expedia, Inc. + * Copyright (C) 2018-2026 Expedia, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -13,7 +13,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package com.expediagroup.apiary.extensions.gluesync.cli; import java.util.ArrayList; @@ -35,11 +34,12 @@ import com.amazonaws.ClientConfiguration; import com.amazonaws.services.glue.AWSGlue; -import com.amazonaws.services.glue.AWSGlueClientBuilder; import com.expediagroup.apiary.extensions.events.metastore.consumer.common.thrift.ThriftHiveClient; import com.expediagroup.apiary.extensions.events.metastore.consumer.common.thrift.ThriftHiveClientFactory; import com.expediagroup.apiary.extensions.gluesync.listener.ApiaryGlueSync; +import com.expediagroup.apiary.extensions.gluesync.listener.GlueClientFactory; +import com.expediagroup.apiary.extensions.gluesync.listener.metrics.MetricService; import com.expediagroup.apiary.extensions.gluesync.listener.service.GlueDatabaseService; import com.expediagroup.apiary.extensions.gluesync.listener.service.GluePartitionService; import com.expediagroup.apiary.extensions.gluesync.listener.service.GlueTableService; @@ -64,19 +64,17 @@ public class GlueSyncCli { private IsIcebergTablePredicate isIcebergTablePredicate; public GlueSyncCli() { + MetricService metricService = new MetricService(); ClientConfiguration clientConfig = new ClientConfiguration(); clientConfig.setRequestTimeout(600000); - AWSGlue glueClient = AWSGlueClientBuilder.standard() - .withRegion(System.getenv("AWS_REGION")) - .withClientConfiguration(clientConfig) - .build(); + String gluePrefix = System.getenv("GLUE_PREFIX"); + AWSGlue glueClient = GlueClientFactory.buildClient(System.getenv("AWS_REGION"), clientConfig, metricService); this.thriftHiveClientFactory = new ThriftHiveClientFactory(); thriftHiveClient = thriftHiveClientFactory.newInstance(THRIFT_CONNECTION_URI, THRIFT_CONNECTION_TIMEOUT); metastoreClient = thriftHiveClient.getMetaStoreClient(); Configuration config = new Configuration(); - this.apiaryGlueSync = new ApiaryGlueSync(config, true); + this.apiaryGlueSync = new ApiaryGlueSync(config, glueClient, gluePrefix, metricService, true); this.isIcebergTablePredicate = new IsIcebergTablePredicate(); - String gluePrefix = System.getenv("GLUE_PREFIX"); this.gluePartitionService = new GluePartitionService(glueClient, gluePrefix); this.glueDatabaseService = new GlueDatabaseService(glueClient, gluePrefix); } diff --git a/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/ApiaryGlueSync.java b/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/ApiaryGlueSync.java index 60a144d6..7c3eac63 100644 --- a/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/ApiaryGlueSync.java +++ b/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/ApiaryGlueSync.java @@ -39,7 +39,6 @@ import org.slf4j.LoggerFactory; import com.amazonaws.services.glue.AWSGlue; -import com.amazonaws.services.glue.AWSGlueClientBuilder; import com.amazonaws.services.glue.model.AlreadyExistsException; import com.amazonaws.services.glue.model.EntityNotFoundException; @@ -78,20 +77,17 @@ public ApiaryGlueSync(Configuration config) { public ApiaryGlueSync(Configuration config, boolean throwExceptions) { super(config); - this.glueClient = AWSGlueClientBuilder.standard().withRegion(System.getenv("AWS_REGION")).build(); + this.metricService = new MetricService(); + this.glueClient = GlueClientFactory.buildClient(System.getenv("AWS_REGION"), metricService); String gluePrefix = System.getenv("GLUE_PREFIX"); this.glueDatabaseService = new GlueDatabaseService(glueClient, gluePrefix); this.gluePartitionService = new GluePartitionService(glueClient, gluePrefix); this.glueTableService = new GlueTableService(glueClient, gluePartitionService, gluePrefix); this.isIcebergPredicate = new IsIcebergTablePredicate(); - this.metricService = new MetricService(); this.throwExceptions = throwExceptions; log.debug("ApiaryGlueSync created"); } - /** - * Just for testing. - */ public ApiaryGlueSync(Configuration config, AWSGlue glueClient, String gluePrefix, MetricService metricService, boolean throwExceptions) { this(config, glueClient, gluePrefix, metricService, throwExceptions, null); diff --git a/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/GlueClientFactory.java b/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/GlueClientFactory.java new file mode 100644 index 00000000..c7433ba0 --- /dev/null +++ b/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/GlueClientFactory.java @@ -0,0 +1,119 @@ +/** + * Copyright (C) 2018-2026 Expedia, Inc. + * + * 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 com.expediagroup.apiary.extensions.gluesync.listener; + +import static com.amazonaws.retry.PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.amazonaws.AmazonServiceException; +import com.amazonaws.ClientConfiguration; +import com.amazonaws.retry.PredefinedRetryPolicies; +import com.amazonaws.retry.RetryPolicy; +import com.amazonaws.services.glue.AWSGlue; +import com.amazonaws.services.glue.AWSGlueClientBuilder; + +import com.expediagroup.apiary.extensions.gluesync.listener.metrics.MetricService; + +/** + * Builds the AWSGlue client with optional CME retry policy and Micrometer metrics. + * + * CME retries are disabled by default to avoid blocking HMS threads. Enable for Dronefly (CLI) via: + * GLUE_RETRY_ENABLED=true + * GLUE_RETRY_MAX_RETRIES=3 (optional, default 3 retries = 4 total calls) + */ +public class GlueClientFactory { + + private static final Logger log = LoggerFactory.getLogger(GlueClientFactory.class); + + static final String ENV_RETRY_ENABLED = "GLUE_RETRY_ENABLED"; + static final String ENV_RETRY_MAX_ATTEMPTS = "GLUE_RETRY_MAX_ATTEMPTS"; + static final int DEFAULT_MAX_RETRIES = 3; + + private GlueClientFactory() {} + + public static AWSGlue buildClient(String region, MetricService metricService) { + return buildClient(region, new ClientConfiguration(), metricService); + } + + public static AWSGlue buildClient(String region, ClientConfiguration config, MetricService metricService) { + if (retryEnabled()) { + int maxRetries = maxRetries(); + log.info("Glue client custom retry policy active: max {} retries per call (covers throttles, 5xx, and CME)", maxRetries); + config.setRetryPolicy(buildRetryPolicy(maxRetries, metricService)); + } else { + log.info("Glue client custom retry policy not active; SDK default retries (throttles/5xx) still apply. Set {}=true to also retry CME.", ENV_RETRY_ENABLED); + } + + AWSGlueClientBuilder builder = AWSGlueClientBuilder.standard() + .withRegion(region) + .withClientConfiguration(config); + + return builder.build(); + } + + static boolean retryEnabled() { + return "true".equalsIgnoreCase(System.getenv(ENV_RETRY_ENABLED)); + } + + static int maxRetries() { + String val = System.getenv(ENV_RETRY_MAX_ATTEMPTS); + if (val != null) { + try { + int n = Integer.parseInt(val.trim()); + if (n > 0) { + return n; + } + } catch (NumberFormatException ignored) { + log.warn("Invalid value for {}: '{}', using default {}", ENV_RETRY_MAX_ATTEMPTS, val, DEFAULT_MAX_RETRIES); + } + } + return DEFAULT_MAX_RETRIES; + } + + static RetryPolicy buildRetryPolicy(int maxRetries, MetricService metricService) { + RetryPolicy.RetryCondition condition = (request, exception, retriesAttempted) -> { + if (DEFAULT_RETRY_CONDITION.shouldRetry(request, exception, retriesAttempted)) { + recordRetry(metricService, exceptionTag(exception)); + return true; + } + if (exception instanceof AmazonServiceException) { + String errorCode = ((AmazonServiceException) exception).getErrorCode(); + if ("ConcurrentModificationException".equals(errorCode)) { + recordRetry(metricService, errorCode); + return true; + } + } + return false; + }; + return new RetryPolicy(condition, PredefinedRetryPolicies.DEFAULT_BACKOFF_STRATEGY, maxRetries, true); + } + + private static String exceptionTag(Exception e) { + if (e instanceof AmazonServiceException) { + String code = ((AmazonServiceException) e).getErrorCode(); + return code != null ? code : e.getClass().getSimpleName(); + } + return e.getClass().getSimpleName(); + } + + private static void recordRetry(MetricService metricService, String exceptionType) { + if (metricService != null) { + metricService.recordGlueRetryAttempt(exceptionType); + } + } +} diff --git a/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/metrics/MetricConstants.java b/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/metrics/MetricConstants.java index f56cccb8..0ebdd00d 100644 --- a/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/metrics/MetricConstants.java +++ b/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/metrics/MetricConstants.java @@ -30,9 +30,12 @@ public class MetricConstants { public static final String LISTENER_EVENT = "glue_listener_event"; public static final String LISTENER_TABLE_RENAME_DURATION = "glue_listener_table_rename_duration"; + public static final String GLUE_RETRY_ATTEMPT = "glue_listener_retry_attempt"; + public static final String TAG_OPERATION = "operation"; public static final String TAG_RESULT = "result"; public static final String TAG_OUTCOME = "outcome"; + public static final String TAG_EXCEPTION = "exception_type"; public static final String RESULT_SUCCESS = "success"; public static final String RESULT_FAILURE = "failure"; diff --git a/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/metrics/MetricService.java b/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/metrics/MetricService.java index 4caa0080..89aef498 100644 --- a/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/metrics/MetricService.java +++ b/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/metrics/MetricService.java @@ -15,7 +15,9 @@ */ package com.expediagroup.apiary.extensions.gluesync.listener.metrics; +import static com.expediagroup.apiary.extensions.gluesync.listener.metrics.MetricConstants.GLUE_RETRY_ATTEMPT; import static com.expediagroup.apiary.extensions.gluesync.listener.metrics.MetricConstants.LISTENER_EVENT; +import static com.expediagroup.apiary.extensions.gluesync.listener.metrics.MetricConstants.TAG_EXCEPTION; import static com.expediagroup.apiary.extensions.gluesync.listener.metrics.MetricConstants.TAG_OPERATION; import static com.expediagroup.apiary.extensions.gluesync.listener.metrics.MetricConstants.TAG_OUTCOME; import static com.expediagroup.apiary.extensions.gluesync.listener.metrics.MetricConstants.TAG_RESULT; @@ -48,6 +50,7 @@ public class MetricService { private final MeterRegistry registry; private final Map metrics; private final Map events = new ConcurrentHashMap<>(); + private final Map retryAttempts = new ConcurrentHashMap<>(); public MetricService(MeterRegistry registry) { this.registry = registry; @@ -137,4 +140,16 @@ public void recordEvent(String operation, String result, String outcome) { log.warn("Unable to record event {} {} {}", operation, result, outcome, e); } } + + public void recordGlueRetryAttempt(String exceptionType) { + try { + retryAttempts.computeIfAbsent(exceptionType, k -> + Counter.builder(GLUE_RETRY_ATTEMPT) + .tags(TAG_EXCEPTION, exceptionType) + .register(registry)) + .increment(); + } catch (Exception e) { + log.warn("Unable to record Glue retry attempt {}", exceptionType, e); + } + } } diff --git a/hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/GlueClientFactoryTest.java b/hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/GlueClientFactoryTest.java new file mode 100644 index 00000000..2fb4dd46 --- /dev/null +++ b/hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/GlueClientFactoryTest.java @@ -0,0 +1,102 @@ +/** + * Copyright (C) 2018-2026 Expedia, Inc. + * + * 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 com.expediagroup.apiary.extensions.gluesync.listener; + +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; + +import org.junit.Before; +import org.junit.Test; + +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; + +import com.amazonaws.AmazonServiceException; +import com.amazonaws.retry.RetryPolicy; + +import com.expediagroup.apiary.extensions.gluesync.listener.metrics.MetricConstants; +import com.expediagroup.apiary.extensions.gluesync.listener.metrics.MetricService; + +public class GlueClientFactoryTest { + + private SimpleMeterRegistry registry; + private MetricService metricService; + + @Before + public void setUp() { + registry = new SimpleMeterRegistry(); + metricService = new MetricService(registry); + } + + @Test + public void defaultMaxRetriesIsThree() { + assertThat(GlueClientFactory.maxRetries(), is(GlueClientFactory.DEFAULT_MAX_RETRIES)); + } + + @Test + public void retryPolicyRetryCountMatchesMaxAttempts() { + RetryPolicy policy = GlueClientFactory.buildRetryPolicy(5, null); + assertThat(policy.getMaxErrorRetry(), is(5)); + } + + @Test + public void concurrentModificationExceptionIsRetried() { + RetryPolicy policy = GlueClientFactory.buildRetryPolicy(3, null); + AmazonServiceException ex = new AmazonServiceException("conflict"); + ex.setErrorCode("ConcurrentModificationException"); + + boolean shouldRetry = policy.getRetryCondition().shouldRetry(null, ex, 0); + assertThat(shouldRetry, is(true)); + } + + @Test + public void unknownServiceExceptionIsNotRetried() { + RetryPolicy policy = GlueClientFactory.buildRetryPolicy(3, null); + AmazonServiceException ex = new AmazonServiceException("bad input"); + ex.setErrorCode("InvalidInputException"); + ex.setStatusCode(400); + + boolean shouldRetry = policy.getRetryCondition().shouldRetry(null, ex, 0); + assertThat(shouldRetry, is(false)); + } + + @Test + public void retryRecordsMetricWhenMetricServiceIsPresent() { + RetryPolicy policy = GlueClientFactory.buildRetryPolicy(3, metricService); + AmazonServiceException ex = new AmazonServiceException("conflict"); + ex.setErrorCode("ConcurrentModificationException"); + + policy.getRetryCondition().shouldRetry(null, ex, 0); + + assertThat(registry.get(MetricConstants.GLUE_RETRY_ATTEMPT) + .tags(MetricConstants.TAG_EXCEPTION, "ConcurrentModificationException") + .counter().count(), is(1.0)); + } + + @Test + public void retryDoesNotThrowWhenMetricServiceIsNull() { + RetryPolicy policy = GlueClientFactory.buildRetryPolicy(3, null); + AmazonServiceException ex = new AmazonServiceException("conflict"); + ex.setErrorCode("ConcurrentModificationException"); + + // should not throw + policy.getRetryCondition().shouldRetry(null, ex, 0); + } + + @Test + public void retryIsDisabledByDefault() { + assertThat(GlueClientFactory.retryEnabled(), is(false)); + } +} diff --git a/hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/metrics/MetricServiceTest.java b/hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/metrics/MetricServiceTest.java index 2464887c..0ab3cbaf 100644 --- a/hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/metrics/MetricServiceTest.java +++ b/hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/metrics/MetricServiceTest.java @@ -91,6 +91,18 @@ public void jmxRegistryExposesCountersAsMBeans() throws Exception { jmxRegistry.close(); } + @Test + public void recordGlueRetryAttemptRegistersTaggedCounter() { + MeterRegistry registry = new SimpleMeterRegistry(); + MetricService metricService = new MetricService(registry); + + metricService.recordGlueRetryAttempt("ConcurrentModificationException"); + + assertThat(registry.get(MetricConstants.GLUE_RETRY_ATTEMPT) + .tags(MetricConstants.TAG_EXCEPTION, "ConcurrentModificationException") + .counter().count(), is(1.0)); + } + @Test public void taggedEventCounterExposesTagsAsJmxKeyProperties() throws Exception { MetricRegistry dropwizardRegistry = new MetricRegistry();