feat: add optional Glue retry with exponential backoff and HTTP-level metrics

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);
   }

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);

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);
+    }
+  }
+}

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";

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<String, Counter> metrics;
   private final Map<String, Counter> events = new ConcurrentHashMap<>();
+  private final Map<String, Counter> 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);
+    }
+  }
 }

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));
+  }
+}

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();