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 cd758fb2..7681b395 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 @@ -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; diff --git a/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/service/GlueTableService.java b/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/service/GlueTableService.java index 2e82fe40..307c657c 100644 --- a/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/service/GlueTableService.java +++ b/hive-event-listeners/apiary-gluesync-listener/src/main/java/com/expediagroup/apiary/extensions/gluesync/listener/service/GlueTableService.java @@ -15,6 +15,10 @@ */ 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; @@ -22,6 +26,8 @@ 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; @@ -29,6 +35,8 @@ 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. + * + *

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 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 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 hmsParams, Map 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; + } } diff --git a/hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/ApiaryGlueSyncTest.java b/hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/ApiaryGlueSyncTest.java index 7abd7b03..458ca2c5 100644 --- a/hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/ApiaryGlueSyncTest.java +++ b/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); diff --git a/hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/service/GlueTableServiceTest.java b/hive-event-listeners/apiary-gluesync-listener/src/test/java/com/expediagroup/apiary/extensions/gluesync/listener/service/GlueTableServiceTest.java new file mode 100644 index 00000000..4cd5f014 --- /dev/null +++ b/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 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 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) null)); + } +}