Add tableProperties to TableAuditEvent

services/tables/src/main/java/com/linkedin/openhouse/tables/audit/TableAuditAspect.java

@@ -17,7 +17,9 @@
 import com.linkedin.openhouse.tables.audit.model.OperationStatus;
 import com.linkedin.openhouse.tables.audit.model.OperationType;
 import com.linkedin.openhouse.tables.audit.model.TableAuditEvent;
+import com.linkedin.openhouse.tables.config.InternalCatalogProperties;
 import java.time.Instant;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
@@ -44,6 +46,8 @@ public class TableAuditAspect {
 
   @Autowired private AuditHandler<TableAuditEvent> tableAuditHandler;
 
+  @Autowired private InternalCatalogProperties internalCatalogProperties;
+
   /**
    * Install the Around advice for getTable() method in OpenHouseTablesApiHandler.
    *
@@ -378,13 +382,20 @@ protected ApiResponse<GetTableResponseBody> auditPutIcebergSnapshots(
             .tableName(tableId)
             .operationType(operationType);
     extractSnapshotInfo(icebergSnapshotRequestBody, eventBuilder);
-    TableAuditEvent event = eventBuilder.build();
     try {
       result = (ApiResponse<GetTableResponseBody>) point.proceed();
+      // Read tableProperties from the response, not the request body: OpenHouse mutates
+      // properties server-side during commit (e.g. openhouse.tableVersion,
+      // openhouse.lastModifiedTime), and the audit event should reflect the committed state.
+      TableAuditEvent event =
+          eventBuilder
+              .tableProperties(filterTableProperties(result.getResponseBody().getTableProperties()))
+              .build();
       buildAndSendEvent(
           event, OperationStatus.SUCCESS, result.getResponseBody().getTableLocation());
     } catch (Throwable t) {
-      buildAndSendEvent(event, OperationStatus.FAILED, null);
+      // On failure there is no committed state to read from, so tableProperties stays null.
+      buildAndSendEvent(eventBuilder.build(), OperationStatus.FAILED, null);
       throw t;
     }
     return result;
@@ -529,6 +540,30 @@ protected ApiResponse<Void> auditUpdateDatabaseAclPolicies(
     return result;
   }
 
+  /**
+   * Narrows the committed table properties down to the configured allowlist ({@code
+   * cluster.iceberg.tables.audit.table-properties-allowlist}). Returns {@code null} when there is
+   * nothing to emit so downstream audit handlers can skip the field entirely. Iterates the
+   * allowlist rather than the source so cost is O(|allowlist|) regardless of source size.
+   */
+  private Map<String, String> filterTableProperties(Map<String, String> source) {
+    if (source == null || source.isEmpty()) {
+      return null;
+    }
+    List<String> allowlist = internalCatalogProperties.getAudit().getTablePropertiesAllowlist();
+    if (allowlist == null || allowlist.isEmpty()) {
+      return null;
+    }
+    Map<String, String> filtered = new HashMap<>();
+    for (String key : allowlist) {
+      String value = source.get(key);
+      if (value != null) {
+        filtered.put(key, value);
+      }
+    }
+    return filtered.isEmpty() ? null : filtered;
+  }
+
   private void buildAndSendEvent(
       TableAuditEvent event, OperationStatus status, String currentTableRoot) {
     TableAuditEvent completeEvent =

services/tables/src/main/java/com/linkedin/openhouse/tables/audit/model/TableAuditEvent.java

@@ -2,6 +2,7 @@
 
 import com.linkedin.openhouse.common.audit.model.BaseAuditEvent;
 import java.time.Instant;
+import java.util.Map;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.EqualsAndHashCode;
@@ -40,4 +41,6 @@ public class TableAuditEvent extends BaseAuditEvent {
   private Long currentSnapshotId;
 
   private Long currentSnapshotTimestampMs;
+
+  private Map<String, String> tableProperties;
 }

services/tables/src/main/java/com/linkedin/openhouse/tables/config/InternalCatalogProperties.java

@@ -1,6 +1,8 @@
 package com.linkedin.openhouse.tables.config;
 
 import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
 import lombok.Getter;
 import lombok.Setter;
 import org.springframework.boot.context.properties.ConfigurationProperties;
@@ -13,11 +15,19 @@ public class InternalCatalogProperties {
 
   private MetadataCache metadataCache = new MetadataCache();
 
+  private Audit audit = new Audit();
+
   @Getter
   @Setter
   public static class MetadataCache {
     private Boolean enabled;
     private Duration ttl;
     private DataSize maxWeight;
   }
+
+  @Getter
+  @Setter
+  public static class Audit {
+    private List<String> tablePropertiesAllowlist = Collections.emptyList();
+  }
 }

services/tables/src/test/java/com/linkedin/openhouse/tables/mock/MockIcebergSnapshotApiHandler.java

@@ -26,9 +26,20 @@ public ApiResponse<GetTableResponseBody> putIcebergSnapshots(
             .responseBody(RequestConstants.TEST_GET_TABLE_RESPONSE_BODY)
             .build();
       case "d200":
+        // Echo the request's table properties on the response so the audit aspect can read
+        // committed state from result.getResponseBody().getTableProperties().
         return ApiResponse.<GetTableResponseBody>builder()
             .httpStatus(HttpStatus.OK)
-            .responseBody(RequestConstants.TEST_GET_TABLE_RESPONSE_BODY)
+            .responseBody(
+                RequestConstants.TEST_GET_TABLE_RESPONSE_BODY
+                    .toBuilder()
+                    .tableProperties(
+                        icebergSnapshotRequestBody.getCreateUpdateTableRequestBody() == null
+                            ? null
+                            : icebergSnapshotRequestBody
+                                .getCreateUpdateTableRequestBody()
+                                .getTableProperties())
+                    .build())
             .build();
       case "d400":
         throw new RequestValidationFailureException();

services/tables/src/test/java/com/linkedin/openhouse/tables/mock/audit/IcebergSnapshotsApiHandlerAuditTest.java

@@ -25,12 +25,18 @@
 import org.springframework.http.MediaType;
 import org.springframework.security.test.context.support.WithMockUser;
 import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.web.servlet.MockMvc;
 import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
 
 @SpringBootTest
 @AutoConfigureMockMvc
 @ContextConfiguration
+@TestPropertySource(
+    properties = {
+      "cluster.iceberg.tables.audit.table-properties-allowlist[0]=openhouse.watermark",
+      "cluster.iceberg.tables.audit.table-properties-allowlist[1]=openhouse.tableType"
+    })
 @WithMockUser(username = "testUser")
 public class IcebergSnapshotsApiHandlerAuditTest {
   @Autowired private MockMvc mvc;
@@ -184,6 +190,40 @@ public void testPutIcebergSnapshotsMainPointsToOlderSnapshot() throws Exception
     assertEquals(1000L, actualEvent.getCurrentSnapshotTimestampMs().longValue());
   }
 
+  @Test
+  public void testPutIcebergSnapshotsFiltersTablePropertiesToAllowlist() throws Exception {
+    Map<String, String> requestProperties = new HashMap<>();
+    requestProperties.put("openhouse.watermark", "100");
+    requestProperties.put("openhouse.tableType", "PRIMARY_TABLE");
+    requestProperties.put("user.custom.key", "v");
+    IcebergSnapshotsRequestBody base = RequestConstants.TEST_ICEBERG_SNAPSHOTS_REQUEST_BODY;
+    IcebergSnapshotsRequestBody requestBody =
+        IcebergSnapshotsRequestBody.builder()
+            .baseTableVersion(base.getBaseTableVersion())
+            .jsonSnapshots(base.getJsonSnapshots())
+            .snapshotRefs(base.getSnapshotRefs())
+            .createUpdateTableRequestBody(
+                base.getCreateUpdateTableRequestBody()
+                    .toBuilder()
+                    .tableProperties(requestProperties)
+                    .build())
+            .build();
+    mvc.perform(
+        MockMvcRequestBuilders.put(
+                String.format(
+                    CURRENT_MAJOR_VERSION_PREFIX
+                        + "/databases/d200/tables/tb1/iceberg/v2/snapshots"))
+            .accept(MediaType.APPLICATION_JSON)
+            .contentType(MediaType.APPLICATION_JSON)
+            .content(requestBody.toJson()));
+    Mockito.verify(tableAuditHandler, atLeastOnce()).audit(argCaptor.capture());
+    TableAuditEvent actualEvent = argCaptor.getValue();
+    Map<String, String> expected = new HashMap<>();
+    expected.put("openhouse.watermark", "100");
+    expected.put("openhouse.tableType", "PRIMARY_TABLE");
+    assertEquals(expected, actualEvent.getTableProperties());
+  }
+
   @Test
   public void testCTASCommitPhase() throws Exception {
     mvc.perform(