fix: optimistic delete guard prevents stale DROP from overwriting renamed Glue table

hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/ApiaryGlueSync.java

@@ -186,7 +186,7 @@ public void onDropTable(DropTableEvent event) throws MetaException {
     }
     Table table = event.getTable();
     try {
-      glueTableService.delete(table);
+      glueTableService.deleteIfUnchanged(table);
       metricService.incrementCounter(MetricConstants.LISTENER_TABLE_SUCCESS);
       metricService.recordEvent(MetricConstants.DROP_TABLE, MetricConstants.RESULT_SUCCESS, "deleted");
     } catch (EntityNotFoundException e) {
@@ -293,7 +293,7 @@ private void doRenameOperation(Table oldTable, Table newTable) {
     long startTime = System.currentTimeMillis();
     glueTableService.create(newTable);
     gluePartitionService.copyPartitions(newTable, gluePartitionService.getPartitions(oldTable));
-    glueTableService.delete(oldTable);
+    glueTableService.deleteIfUnchanged(oldTable);
     metricService.incrementCounter(MetricConstants.LISTENER_TABLE_SUCCESS);
     metricService.recordEvent(MetricConstants.ALTER_TABLE, MetricConstants.RESULT_SUCCESS, "renamed");
     long duration = System.currentTimeMillis() - startTime;

hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/service/GlueTableService.java

@@ -15,20 +15,28 @@
  */
 package com.expediagroup.apiary.extensions.gluesync.listener.service;
 
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+
 import org.apache.hadoop.hive.metastore.api.Table;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.amazonaws.services.glue.AWSGlue;
 import com.amazonaws.services.glue.model.CreateTableRequest;
 import com.amazonaws.services.glue.model.DeleteTableRequest;
+import com.amazonaws.services.glue.model.EntityNotFoundException;
+import com.amazonaws.services.glue.model.GetTableRequest;
 import com.amazonaws.services.glue.model.InvalidInputException;
 import com.amazonaws.services.glue.model.TableInput;
 import com.amazonaws.services.glue.model.UpdateTableRequest;
 import com.amazonaws.services.glue.model.ValidationException;
 
 public class GlueTableService {
   private static final Logger log = LoggerFactory.getLogger(GlueTableService.class);
+  // HMS parameters checked before a delete to detect if the table slot was overwritten by a concurrent operation
+  private static final String[] IDENTITY_PARAMS = { "transient_lastDdlTime", "metadata_location" };
   public static final String APIARY_GLUESYNC_SKIP_ARCHIVE_TABLE_PARAM = "apiary.gluesync.skipArchive";
 
   private final AWSGlue glueClient;
@@ -92,4 +100,50 @@ public void delete(Table table) {
     glueClient.deleteTable(deleteTableRequest);
     log.debug(table + " table deleted from glue catalog");
   }
+
+  /**
+   * Deletes the Glue table only if {@link #IDENTITY_PARAMS} match the current Glue state, guarding
+   * against a race condition in multi-partition Kafka deployments where a concurrent rename can
+   * overwrite the table slot before a stale drop event is processed. Safe in direct-HMS deployments
+   * because HMS does not update {@code transient_lastDdlTime} on DROP, so the event value always
+   * matches the last-synced Glue value.
+   *
+   * <p>Caveat: if a prior sync failed (e.g. transient Glue error), the params will diverge and the
+   * delete will be skipped, leaving an orphan in Glue. An orphan is intentionally preferred over
+   * incorrectly deleting a table that has since been overwritten by a rename.
+   */
+  public void deleteIfUnchanged(Table table) {
+    Map<String, String> hmsParams = table.getParameters() != null ? table.getParameters() : Collections.emptyMap();
+    com.amazonaws.services.glue.model.Table glueTable;
+    try {
+      glueTable = glueClient
+          .getTable(new GetTableRequest()
+              .withDatabaseName(transformer.glueDbName(table))
+              .withName(table.getTableName()))
+          .getTable();
+    } catch (EntityNotFoundException e) {
+      log.info("{}.{} table not found in glue catalog, nothing to delete", table.getDbName(), table.getTableName());
+      return;
+    }
+    Map<String, String> glueParams = glueTable.getParameters() != null ? glueTable.getParameters() : Collections.emptyMap();
+
+    for (String key : IDENTITY_PARAMS) {
+      if (paramChanged(key, hmsParams, glueParams, table)) {
+        return;
+      }
+    }
+
+    delete(table);
+  }
+
+  private boolean paramChanged(String key, Map<String, String> hmsParams, Map<String, String> glueParams, Table table) {
+    String hmsValue = hmsParams.get(key);
+    String glueValue = glueParams.get(key);
+    if (!Objects.equals(hmsValue, glueValue)) {
+      log.warn("Skipping delete of {}.{}: {} changed (expected={}, found={})",
+          table.getDbName(), table.getTableName(), key, hmsValue, glueValue);
+      return true;
+    }
+    return false;
+  }
 }

hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/ApiaryGlueSyncTest.java

@@ -88,6 +88,7 @@
 import com.amazonaws.services.glue.model.GetDatabaseRequest;
 import com.amazonaws.services.glue.model.GetDatabaseResult;
 import com.amazonaws.services.glue.model.GetPartitionsResult;
+import com.amazonaws.services.glue.model.GetTableResult;
 import com.amazonaws.services.glue.model.InvalidInputException;
 import com.amazonaws.services.glue.model.OperationTimeoutException;
 import com.amazonaws.services.glue.model.PartitionInput;
@@ -471,6 +472,7 @@ public void onAlterHiveTable_RenameTable() throws MetaException {
 
     when(glueClient.getPartitions(any())).thenReturn(new GetPartitionsResult().withPartitions(
         new com.amazonaws.services.glue.model.Partition().withValues("part1Value", "part2Value")));
+    when(glueClient.getTable(any())).thenReturn(new GetTableResult().withTable(new com.amazonaws.services.glue.model.Table()));
 
     glueSync.onAlterTable(event);
 
@@ -703,6 +705,7 @@ public void onDropTable() throws MetaException {
     DropTableEvent event = mock(DropTableEvent.class);
     when(event.getStatus()).thenReturn(true);
     when(event.getTable()).thenReturn(simpleHiveTable(simpleSchema(), simplePartitioning()));
+    when(glueClient.getTable(any())).thenReturn(new GetTableResult().withTable(new com.amazonaws.services.glue.model.Table()));
 
     glueSync.onDropTable(event);
 
@@ -718,6 +721,7 @@ public void onDropTableThatDoesntExistInGlue() throws MetaException {
     DropTableEvent event = mock(DropTableEvent.class);
     when(event.getStatus()).thenReturn(true);
     when(event.getTable()).thenReturn(simpleHiveTable(simpleSchema(), simplePartitioning()));
+    when(glueClient.getTable(any())).thenReturn(new GetTableResult().withTable(new com.amazonaws.services.glue.model.Table()));
     when(glueClient.deleteTable(any())).thenThrow(new EntityNotFoundException(""));
 
     glueSync.onDropTable(event);

hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/service/GlueTableServiceTest.java

@@ -0,0 +1,197 @@
+/**
+ * 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.service;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import com.amazonaws.services.glue.AWSGlue;
+import com.amazonaws.services.glue.model.DeleteTableRequest;
+import com.amazonaws.services.glue.model.EntityNotFoundException;
+import com.amazonaws.services.glue.model.GetTableRequest;
+import com.amazonaws.services.glue.model.GetTableResult;
+
+@RunWith(MockitoJUnitRunner.class)
+public class GlueTableServiceTest {
+
+  private static final String DB_NAME = "test_db";
+  private static final String TABLE_NAME = "test_table";
+  private static final String LOCATION = "s3://bucket/test_table";
+  private static final String LAST_DDL_TIME = "1751385266";
+
+  @Mock
+  private AWSGlue glueClient;
+  @Mock
+  private GluePartitionService gluePartitionService;
+
+  private GlueTableService service;
+
+  @Before
+  public void setUp() {
+    service = new GlueTableService(glueClient, gluePartitionService, null);
+  }
+
+  @Test
+  public void deleteIfUnchanged_deletesWhenAllPropertiesMatch() {
+    when(glueClient.getTable(any(GetTableRequest.class))).thenReturn(glueTableResult(LOCATION, LAST_DDL_TIME, null));
+
+    service.deleteIfUnchanged(hmsTable(LOCATION, LAST_DDL_TIME, null));
+
+    verify(glueClient).deleteTable(any(DeleteTableRequest.class));
+  }
+
+  @Test
+  public void deleteIfUnchanged_skipsWhenLastDdlTimeChanged() {
+    when(glueClient.getTable(any(GetTableRequest.class))).thenReturn(glueTableResult(LOCATION, "1751385999", null));
+
+    service.deleteIfUnchanged(hmsTable(LOCATION, LAST_DDL_TIME, null));
+
+    verify(glueClient, never()).deleteTable(any(DeleteTableRequest.class));
+  }
+
+  @Test
+  public void deleteIfUnchanged_skipsWhenHiveTableReplacedByIceberg() {
+    // Hive DROP event has no metadata_location; Glue now holds an Iceberg table
+    when(glueClient.getTable(any(GetTableRequest.class)))
+        .thenReturn(glueTableResult(LOCATION, LAST_DDL_TIME, "s3://bucket/test_table/metadata/v1.metadata.json"));
+
+    service.deleteIfUnchanged(hmsTable(LOCATION, LAST_DDL_TIME, null));
+
+    verify(glueClient, never()).deleteTable(any(DeleteTableRequest.class));
+  }
+
+  @Test
+  public void deleteIfUnchanged_skipsWhenMetadataLocationDiverges() {
+    when(glueClient.getTable(any(GetTableRequest.class)))
+        .thenReturn(glueTableResult(LOCATION, LAST_DDL_TIME, "s3://bucket/test_table/metadata/v2.metadata.json"));
+
+    service.deleteIfUnchanged(hmsTable(LOCATION, LAST_DDL_TIME, "s3://bucket/test_table/metadata/v1.metadata.json"));
+
+    verify(glueClient, never()).deleteTable(any(DeleteTableRequest.class));
+  }
+
+  @Test
+  public void deleteIfUnchanged_skipsWhenHmsHasMetadataLocationButGlueDoesNot() {
+    when(glueClient.getTable(any(GetTableRequest.class))).thenReturn(glueTableResult(LOCATION, LAST_DDL_TIME, null));
+
+    service.deleteIfUnchanged(hmsTable(LOCATION, LAST_DDL_TIME, "s3://bucket/test_table/metadata/v1.metadata.json"));
+
+    verify(glueClient, never()).deleteTable(any(DeleteTableRequest.class));
+  }
+
+  @Test
+  public void deleteIfUnchanged_skipsWhenTableNotFoundInGlue() {
+    when(glueClient.getTable(any(GetTableRequest.class))).thenThrow(new EntityNotFoundException("not found"));
+
+    service.deleteIfUnchanged(hmsTable(LOCATION, LAST_DDL_TIME, null));
+
+    verify(glueClient, never()).deleteTable(any(DeleteTableRequest.class));
+  }
+
+  @Test
+  public void deleteIfUnchanged_deletesWhenBothParamMapsAreNull() {
+    when(glueClient.getTable(any(GetTableRequest.class))).thenReturn(glueTableResultNullParams());
+
+    Table table = hmsTable(LOCATION, null, null);
+    table.setParameters(null);
+    service.deleteIfUnchanged(table);
+
+    verify(glueClient).deleteTable(any(DeleteTableRequest.class));
+  }
+
+  @Test
+  public void deleteIfUnchanged_skipsWhenHmsParamsNullButGlueHasParams() {
+    when(glueClient.getTable(any(GetTableRequest.class))).thenReturn(glueTableResult(LOCATION, LAST_DDL_TIME, null));
+
+    Table table = hmsTable(LOCATION, null, null);
+    table.setParameters(null);
+    service.deleteIfUnchanged(table);
+
+    verify(glueClient, never()).deleteTable(any(DeleteTableRequest.class));
+  }
+
+  @Test
+  public void deleteIfUnchanged_skipsWhenGlueParamsNullButHmsHasParams() {
+    when(glueClient.getTable(any(GetTableRequest.class))).thenReturn(glueTableResultNullParams());
+
+    service.deleteIfUnchanged(hmsTable(LOCATION, LAST_DDL_TIME, null));
+
+    verify(glueClient, never()).deleteTable(any(DeleteTableRequest.class));
+  }
+
+  private Table hmsTable(String location, String lastDdlTime, String metadataLocation) {
+    Table table = new Table();
+    table.setTableName(TABLE_NAME);
+    table.setDbName(DB_NAME);
+
+    StorageDescriptor sd = new StorageDescriptor();
+    sd.setLocation(location);
+    sd.setSerdeInfo(new SerDeInfo());
+    table.setSd(sd);
+
+    Map<String, String> params = new HashMap<>();
+    if (lastDdlTime != null) {
+      params.put("transient_lastDdlTime", lastDdlTime);
+    }
+    if (metadataLocation != null) {
+      params.put("metadata_location", metadataLocation);
+    }
+    table.setParameters(params);
+    return table;
+  }
+
+  private GetTableResult glueTableResult(String location, String lastDdlTime, String metadataLocation) {
+    com.amazonaws.services.glue.model.StorageDescriptor sd = new com.amazonaws.services.glue.model.StorageDescriptor()
+        .withLocation(location);
+
+    Map<String, String> params = new HashMap<>();
+    if (lastDdlTime != null) {
+      params.put("transient_lastDdlTime", lastDdlTime);
+    }
+    if (metadataLocation != null) {
+      params.put("metadata_location", metadataLocation);
+    }
+
+    return new GetTableResult().withTable(
+        new com.amazonaws.services.glue.model.Table()
+            .withName(TABLE_NAME)
+            .withDatabaseName(DB_NAME)
+            .withStorageDescriptor(sd)
+            .withParameters(params));
+  }
+
+  private GetTableResult glueTableResultNullParams() {
+    return new GetTableResult().withTable(
+        new com.amazonaws.services.glue.model.Table()
+            .withName(TABLE_NAME)
+            .withDatabaseName(DB_NAME)
+            .withParameters((Map<String, String>) null));
+  }
+}