jamespfaulkner · Oct 8, 2026 · Oct 7, 2026 · Oct 7, 2026 · Oct 7, 2026 · Oct 7, 2026
@@ -3,6 +3,10 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## 8.2.6 - 2026-10-07
+### Added
+- `apiary-gluesync-listener`: new `GLUE_SEND_VERSION_ID` environment variable (default `false`). When `true`, the listener sends the current Glue `VersionId` on `UpdateTable` requests (all table types) so the emitted CloudTrail event carries `requestParameters.versionId`, which lets consumers of the Glue/CloudTrail event stream identify and fetch the pre-update table version for `ALTER_TABLE` events. Glue enforces the `VersionId` as an optimistic-concurrency token, so if a concurrent writer causes `ConcurrentModificationException`, or the current `VersionId` cannot be read (any Glue failure other than the table not existing), the listener falls back to a single unconditional update rather than failing the sync — matching its last-write-wins behaviour when the feature is off. The flag does not change archiving behaviour (`GLUE_SKIP_ARCHIVE` / `apiary.gluesync.skipArchive` still govern `SkipArchive`), but note the carried `versionId` only resolves downstream while the previous version is archived. New `version_conflict_fallback` and `version_unavailable_fallback` values can appear on the `glue_listener_event` counter's `outcome` tag.
+
 ## 8.2.5 - 2026-08-31
 ### Added
 - `kafka-metastore-receiver`: `KafkaMessageReaderBuilder.withKeyDeserializer(String)` to configure the record key deserializer. The default remains `LongDeserializer`, matching the key written by the Apiary Hive Metastore listener, but topics populated by another producer can now be read without a `SerializationException`. The key deserializer could not previously be set, because properties passed to `withConsumerProperties` do not override the builder's own defaults.

@@ -17,6 +17,14 @@ The GlueSync listener can be configured by setting the following System Environm
 GLUE_PREFIX|No|Prefix added to Glue databases to handle database name collisions when synchronizing multiple metastores to the Glue catalog.
 ENABLE_HIVE_TO_GLUE_RENAME_OPERATION|No|Set to true in case you would like to enable Hive table renames when syncing into Glue. Default value is false.
 GLUE_SKIP_ARCHIVE|No|Default value applied to `SkipArchive` on AWS Glue `UpdateTable` requests when a table does not set the `apiary.gluesync.skipArchive` property. Accepts `true` or `false` (case-insensitive). When unset, the built-in default (`true`) is used. Any other value causes the listener to fail on startup.
+GLUE_SEND_VERSION_ID|No|When `true`, the listener sends the current Glue `VersionId` on `UpdateTable` requests, so the emitted Glue/CloudTrail event carries it (see [Table update VersionId](#table-update-versionid)). Accepts `true` or `false` (case-insensitive). Default is `false`. Any other value causes the listener to fail on startup.
+
+## Table update VersionId
+When `GLUE_SEND_VERSION_ID=true`, the listener reads the current Glue `VersionId` (via `GetTable`) and passes it on every `UpdateTable`. This makes AWS log `requestParameters.versionId` on the emitted CloudTrail `UpdateTable` event — the value is the version the table was at immediately *before* the update, which lets consumers of the Glue/CloudTrail event stream identify and fetch the pre-update table state (via `GetTableVersion`) for `ALTER_TABLE` diffs, rather than scanning version history to compare properties.
+
+AWS Glue treats `VersionId` as an optimistic-concurrency (compare-and-swap) token, so a concurrent writer causes the `UpdateTable` to fail with `ConcurrentModificationException`. In that case the listener retries the update once without a `VersionId`, so the sync still succeeds — that one event just will not carry a `VersionId`. Likewise, if the current `VersionId` cannot be read at all (any Glue failure other than the table not existing), the listener updates unconditionally rather than failing the sync.
+
+For the carried `VersionId` to resolve to fetchable content downstream, the previous version must be archived — i.e. `GLUE_SKIP_ARCHIVE` / `apiary.gluesync.skipArchive` must not skip archiving for the tables concerned. This flag does not change the archiving behaviour (see below); set the skip-archive configuration accordingly.
 
 ## Table update SkipArchive
 [AWS default](https://docs.aws.amazon.com/glue/latest/webapi/API_UpdateTable.html#Glue-UpdateTable-request-SkipArchive) is to archive the table on every update. With Iceberg tables this can lead to a lot of table versions. In Glue you can only have a certain limit of the number of versions and you'll get exceptions when trying to update a table once you hit that limit. Manual version removal through AWS api is then needed. To counter this the listener defaults to `skipArchive=true`, so it does *not* make an archive of the table when updating.

@@ -45,6 +45,7 @@
 
 import com.expediagroup.apiary.extensions.gluesync.listener.metrics.MetricConstants;
 import com.expediagroup.apiary.extensions.gluesync.listener.metrics.MetricService;
+import com.expediagroup.apiary.extensions.gluesync.listener.service.DeleteOutcome;
 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;
@@ -109,11 +110,20 @@ public ApiaryGlueSync(Configuration config, AWSGlue glueClient, String gluePrefi
    */
   public ApiaryGlueSync(Configuration config, AWSGlue glueClient, String gluePrefix, MetricService metricService,
       boolean throwExceptions, Boolean defaultSkipArchive) {
+    this(config, glueClient, gluePrefix, metricService, throwExceptions, defaultSkipArchive, false);
+  }
+
+  /**
+   * Just for testing. Additionally allows injecting the {@code GLUE_SEND_VERSION_ID} flag that
+   * would otherwise be read from the environment.
+   */
+  public ApiaryGlueSync(Configuration config, AWSGlue glueClient, String gluePrefix, MetricService metricService,
+      boolean throwExceptions, Boolean defaultSkipArchive, boolean sendVersionId) {
     super(config);
     this.glueClient = glueClient;
     this.glueDatabaseService = new GlueDatabaseService(glueClient, gluePrefix);
     this.gluePartitionService = new GluePartitionService(glueClient, gluePrefix, defaultSkipArchive);
-    this.glueTableService = new GlueTableService(glueClient, gluePartitionService, gluePrefix);
+    this.glueTableService = new GlueTableService(glueClient, gluePartitionService, gluePrefix, sendVersionId);
     this.isIcebergPredicate = new IsIcebergTablePredicate();
     this.metricService = metricService;
     this.throwExceptions = throwExceptions;
@@ -191,7 +201,7 @@ public void onDropTable(DropTableEvent event) throws MetaException {
     }
     Table table = event.getTable();
     try {
-      GlueTableService.DeleteOutcome outcome = glueTableService.deleteIfUnchanged(table);
+      DeleteOutcome outcome = glueTableService.deleteIfUnchanged(table);
       metricService.incrementCounter(MetricConstants.LISTENER_TABLE_SUCCESS);
       metricService.recordEvent(MetricConstants.DROP_TABLE, MetricConstants.RESULT_SUCCESS, outcome.metricOutcome());
     } catch (Exception e) {
@@ -249,15 +259,13 @@ private String createOrUpdateTable(Table table) {
       return MetricConstants.OUTCOME_CREATED;
     } catch (AlreadyExistsException e) {
       log.info("{} table already exists in glue, updating....", table.getTableName());
-      glueTableService.update(table);
-      return MetricConstants.OUTCOME_UPDATED;
+      return glueTableService.update(table).metricOutcome();
     }
   }
 
   private String updateOrCreateTable(Table table) {
     try {
-      glueTableService.update(table);
-      return MetricConstants.OUTCOME_UPDATED;
+      return glueTableService.update(table).metricOutcome();
     } catch (EntityNotFoundException e) {
       log.info("{} table doesn't exist in glue, creating....", table.getTableName());
       glueTableService.create(table);
@@ -292,7 +300,7 @@ private void doRenameOperation(Table oldTable, Table newTable) {
     long startTime = System.currentTimeMillis();
     glueTableService.create(newTable);
     gluePartitionService.copyPartitions(newTable, gluePartitionService.getPartitions(oldTable));
-    GlueTableService.DeleteOutcome outcome = glueTableService.deleteIfUnchanged(oldTable);
+    DeleteOutcome outcome = glueTableService.deleteIfUnchanged(oldTable);
     long duration = System.currentTimeMillis() - startTime;
     metricService.incrementCounter(MetricConstants.LISTENER_TABLE_SUCCESS);
     metricService.recordEvent(MetricConstants.RENAME_TABLE, MetricConstants.RESULT_SUCCESS, outcome.metricOutcome());

@@ -0,0 +1,45 @@
+/**
+ * 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;
+
+/**
+ * Outcome of a {@link GlueTableService#deleteIfUnchanged} call, used by callers to record the
+ * appropriate metric.
+ * <ul>
+ *   <li>{@code DELETED} — the Glue table was deleted.</li>
+ *   <li>{@code SKIPPED} — the delete was suppressed because identity params diverged, indicating
+ *       the Glue slot was overwritten by a concurrent operation.</li>
+ *   <li>{@code NOT_FOUND} — the table did not exist in Glue when checked; nothing to delete.</li>
+ *   <li>{@code DELETED_CONCURRENTLY} — the table existed at guard-check time but was deleted by a
+ *       concurrent operation before the delete call completed (TOCTOU race).</li>
+ * </ul>
+ */
+public enum DeleteOutcome {
+  DELETED("deleted"),
+  SKIPPED("delete_skipped"),
+  NOT_FOUND("not_found"),
+  DELETED_CONCURRENTLY("concurrent_delete");
+
+  private final String metricOutcome;
+
+  DeleteOutcome(String metricOutcome) {
+    this.metricOutcome = metricOutcome;
+  }
+
+  public String metricOutcome() {
+    return metricOutcome;
+  }
+}
@@ -23,7 +23,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.amazonaws.AmazonClientException;
 import com.amazonaws.services.glue.AWSGlue;
+import com.amazonaws.services.glue.model.ConcurrentModificationException;
 import com.amazonaws.services.glue.model.CreateTableRequest;
 import com.amazonaws.services.glue.model.DeleteTableRequest;
 import com.amazonaws.services.glue.model.EntityNotFoundException;
@@ -35,51 +37,55 @@
 
 public class GlueTableService {
 
-  /**
-   * Outcome of a {@link #deleteIfUnchanged} call, used by callers to record the appropriate metric.
-   * <ul>
-   *   <li>{@code DELETED} — the Glue table was deleted.</li>
-   *   <li>{@code SKIPPED} — the delete was suppressed because identity params diverged, indicating
-   *       the Glue slot was overwritten by a concurrent operation.</li>
-   *   <li>{@code NOT_FOUND} — the table did not exist in Glue when checked; nothing to delete.</li>
-   *   <li>{@code DELETED_CONCURRENTLY} — the table existed at guard-check time but was deleted by a
-   *       concurrent operation before the delete call completed (TOCTOU race).</li>
-   * </ul>
-   */
-  public enum DeleteOutcome {
-    DELETED("deleted"),
-    SKIPPED("delete_skipped"),
-    NOT_FOUND("not_found"),
-    DELETED_CONCURRENTLY("concurrent_delete");
-
-    private final String metricOutcome;
-
-    DeleteOutcome(String metricOutcome) {
-      this.metricOutcome = metricOutcome;
-    }
-
-    public String metricOutcome() {
-      return metricOutcome;
-    }
-  }
-
   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";
+  public static final String GLUE_SEND_VERSION_ID_ENV = "GLUE_SEND_VERSION_ID";
 
   private final AWSGlue glueClient;
   private final HiveToGlueTransformer transformer;
   private final GlueMetadataStringCleaner cleaner = new GlueMetadataStringCleaner();
   private final GluePartitionService gluePartitionService;
+  private final boolean sendVersionId;
 
   public GlueTableService(AWSGlue glueClient, GluePartitionService gluePartitionService, String gluePrefix) {
+    this(glueClient, gluePartitionService, gluePrefix, parseSendVersionIdFromEnv());
+  }
+
+  public GlueTableService(AWSGlue glueClient, GluePartitionService gluePartitionService, String gluePrefix,
+      boolean sendVersionId) {
     this.glueClient = glueClient;
     this.transformer = new HiveToGlueTransformer(gluePrefix);
     this.gluePartitionService = gluePartitionService;
+    this.sendVersionId = sendVersionId;
     log.debug("ApiaryGlueSync created");
   }
 
+  /**
+   * Reads the {@value #GLUE_SEND_VERSION_ID_ENV} environment variable and returns its boolean
+   * value. When unset or empty, {@code false} is returned (the feature is off by default). Any
+   * value other than {@code "true"}/{@code "false"} (case-insensitive) throws at startup.
+   */
+  static boolean parseSendVersionIdFromEnv() {
+    return parseSendVersionId(System.getenv(GLUE_SEND_VERSION_ID_ENV));
+  }
+
+  static boolean parseSendVersionId(String value) {
+    if (value == null || value.isEmpty()) {
+      return false;
+    }
+    if ("true".equalsIgnoreCase(value)) {
+      return true;
+    }
+    if ("false".equalsIgnoreCase(value)) {
+      return false;
+    }
+    throw new IllegalArgumentException(
+        "Invalid value for environment variable " + GLUE_SEND_VERSION_ID_ENV
+            + ": '" + value + "'. Expected 'true' or 'false'.");
+  }
+
   public void create(Table table) {
     CreateTableRequest createTableRequest = new CreateTableRequest()
         .withTableInput(transformer.transformTable(table))
@@ -95,13 +101,61 @@ public void create(Table table) {
     }
   }
 
-  public void update(Table table) {
-    boolean skipArchive = gluePartitionService.shouldSkipArchive(table);
-
+  public UpdateOutcome update(Table table) {
     UpdateTableRequest updateTableRequest = new UpdateTableRequest()
-        .withSkipArchive(skipArchive)
+        .withSkipArchive(gluePartitionService.shouldSkipArchive(table))
         .withTableInput(transformer.transformTable(table))
         .withDatabaseName(transformer.glueDbName(table));
+
+    if (!sendVersionId) {
+      doUpdate(updateTableRequest, table);
+      return UpdateOutcome.UPDATED;
+    }
+    return updateWithOptimisticLock(updateTableRequest, table);
+  }
+
+  /**
+   * Updates the table carrying Glue's optimistic-concurrency {@code versionId}, so AWS records
+   * {@code requestParameters.versionId} (the version immediately before this update) on the emitted
+   * CloudTrail event. The versionId is best-effort: if it cannot be read, or Glue rejects the write
+   * with {@code ConcurrentModificationException} because a concurrent writer bumped the version, the
+   * table is updated unconditionally (no {@code versionId}) so the sync still succeeds — matching
+   * the listener's last-write-wins behaviour when the feature is off. Only that write loses its
+   * {@code versionId}; the outcome records which fallback was taken.
+   */
+  private UpdateOutcome updateWithOptimisticLock(UpdateTableRequest updateTableRequest, Table table) {
+    String versionId = readVersionId(table);
+    if (versionId == null) {
+      doUpdate(updateTableRequest, table);
+      return UpdateOutcome.VERSION_UNAVAILABLE_FALLBACK;
+    }
+    updateTableRequest.setVersionId(versionId);
+    try {
+      doUpdate(updateTableRequest, table);
+      return UpdateOutcome.UPDATED;
+    } catch (ConcurrentModificationException e) {
+      log.warn("Concurrent modification updating {}.{} in glue; updating unconditionally without versionId",
+          table.getDbName(), table.getTableName());
+      updateTableRequest.setVersionId(null);
+      doUpdate(updateTableRequest, table);
+      return UpdateOutcome.VERSION_CONFLICT_FALLBACK;
+    }
+  }
+
+  /**
+   * Whether to set the optimistic-concurrency {@code versionId} on this table's Glue
+   * {@code UpdateTable}. Only non-Iceberg (Hive) tables need it: setting it makes AWS record
+   * {@code requestParameters.versionId} on the emitted CloudTrail {@code UpdateTable} event, which
+   * consumers of that event use to identify the pre-update table version. Iceberg tables are
+   * identified by {@code metadata_location}, so they skip the extra {@code GetTable} and the
+   * optimistic-locking overhead.
+   */
+  /**
+   * Issues the Glue UpdateTable, retrying once with cleaned-up comments if Glue rejects the input
+   * with {@link ValidationException}/{@link InvalidInputException}. Any {@code versionId} set on
+   * the request is preserved across the retry.
+   */
+  private void doUpdate(UpdateTableRequest updateTableRequest, Table table) {
     try {
       glueClient.updateTable(updateTableRequest);
       log.debug(table + " table updated in glue catalog");
@@ -113,6 +167,30 @@ public void update(Table table) {
     }
   }
 
+  /**
+   * Reads the current Glue {@code VersionId} for the table. Returns {@code null} if it cannot be
+   * read (any AWS failure other than the table not existing), so the caller can degrade to an
+   * unconditional update rather than failing the sync for a best-effort versionId. An
+   * {@link EntityNotFoundException} is allowed to propagate so the caller can create the missing
+   * table.
+   */
+  private String readVersionId(Table table) {
+    try {
+      return glueClient
+          .getTable(new GetTableRequest()
+              .withDatabaseName(transformer.glueDbName(table))
+              .withName(table.getTableName()))
+          .getTable()
+          .getVersionId();
+    } catch (EntityNotFoundException e) {
+      throw e;
+    } catch (AmazonClientException e) {
+      log.warn("Could not read current Glue versionId for {}.{}; updating without it",
+          table.getDbName(), table.getTableName(), e);
+      return null;
+    }
+  }
+
   private TableInput cleanUpTable(TableInput table) {
     log.debug("Cleaning up table comments manually on {} to resolve validation", table);
     long startTime = System.currentTimeMillis();