DO NOT MERGE: Idea on Redacting of the Sensitive Options

runtime/defaults/src/main/resources/application.properties

@@ -144,6 +144,7 @@ polaris.event-listener.type=no-op
 # polaris.event-listener.aws-cloudwatch.log-stream=polaris-cloudwatch-default-stream
 # polaris.event-listener.aws-cloudwatch.region=us-east-1
 # polaris.event-listener.aws-cloudwatch.synchronous-mode=false
+# polaris.event-listener.aws-cloudwatch.event-types= // the absence of this property would result in processing all Polaris event types.
 
 polaris.log.request-id-header-name=Polaris-Request-Id
 # polaris.log.mdc.aid=polaris

runtime/service/src/main/java/org/apache/polaris/service/config/PolarisIcebergObjectMapperCustomizer.java

@@ -30,7 +30,12 @@
 import io.smallrye.config.WithConverter;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.rest.RESTSerializers;
+import org.apache.polaris.service.events.PolarisEvent;
+import org.apache.polaris.service.events.json.mixins.IcebergMixins;
+import org.apache.polaris.service.events.json.mixins.PolarisEventBaseMixin;
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -56,6 +61,10 @@ public void customize(ObjectMapper objectMapper) {
     objectMapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY);
     objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
     objectMapper.setPropertyNamingStrategy(new PropertyNamingStrategies.KebabCaseStrategy());
+    objectMapper.addMixIn(PolarisEvent.class, PolarisEventBaseMixin.class);
+    objectMapper.addMixIn(TableIdentifier.class, IcebergMixins.TableIdentifierMixin.class);
+    objectMapper.addMixIn(Namespace.class, IcebergMixins.NamespaceMixin.class);
+
     RESTSerializers.registerAll(objectMapper);
     Serializers.registerSerializers(objectMapper);
     objectMapper

runtime/service/src/main/java/org/apache/polaris/service/events/json/FieldRedactionSerializerModifier.java

@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.polaris.service.events.json;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.BeanDescription;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializationConfig;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.ser.BeanPropertyWriter;
+import com.fasterxml.jackson.databind.ser.BeanSerializerModifier;
+import java.io.IOException;
+import java.util.List;
+import java.util.Set;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+/**
+ * A Jackson BeanSerializerModifier that redacts fields based on configured patterns.
+ *
+ * <p>This modifier intercepts serialization of bean properties and replaces values with a redaction
+ * marker for fields that match the configured patterns. It supports:
+ *
+ * <ul>
+ *   <li>Simple field names (e.g., "password", "secret")
+ *   <li>Nested field paths using dot notation (e.g., "properties.api-key")
+ *   <li>Wildcard patterns (e.g., "*.secret", "credentials.*")
+ * </ul>
+ */
+public class FieldRedactionSerializerModifier extends BeanSerializerModifier {
+
+  private final List<Pattern> fieldPatterns;
+
+  /**
+   * Creates a new FieldRedactionSerializerModifier with the specified field patterns.
+   *
+   * @param fieldNames set of field names or patterns to redact
+   */
+  public FieldRedactionSerializerModifier(Set<String> fieldNames) {
+    this.fieldPatterns =
+        fieldNames.stream()
+            .map(FieldRedactionSerializerModifier::convertToRegexPattern)
+            .map(Pattern::compile)
+            .collect(Collectors.toList());
+  }
+
+  /**
+   * Converts a field pattern (which may contain wildcards) to a regex pattern.
+   *
+   * @param pattern the field pattern (e.g., "*.secret", "credentials.*", "exact-field")
+   * @return a regex pattern string
+   */
+  private static String convertToRegexPattern(String pattern) {
+    // Escape special regex characters except for our wildcard (*)
+    String escaped = pattern.replace(".", "\\.").replace("*", ".*");
+    // Match the entire field path
+    return "^" + escaped + "$";
+  }
+
+  @Override
+  public List<BeanPropertyWriter> changeProperties(
+      SerializationConfig config,
+      BeanDescription beanDesc,
+      List<BeanPropertyWriter> beanProperties) {
+
+    if (fieldPatterns.isEmpty()) {
+      return beanProperties;
+    }
+
+    return beanProperties.stream()
+        .map(writer -> shouldRedactField(writer.getName()) ? createRedactingWriter(writer) : writer)
+        .collect(Collectors.toList());
+  }
+
+  /**
+   * Checks if a field name matches any of the configured redaction patterns.
+   *
+   * @param fieldName the field name to check
+   * @return true if the field should be redacted
+   */
+  private boolean shouldRedactField(String fieldName) {
+    return fieldPatterns.stream().anyMatch(pattern -> pattern.matcher(fieldName).matches());
+  }
+
+  /**
+   * Creates a new BeanPropertyWriter that redacts the field value.
+   *
+   * @param original the original BeanPropertyWriter
+   * @return a new BeanPropertyWriter that redacts the value
+   */
+  private BeanPropertyWriter createRedactingWriter(BeanPropertyWriter original) {
+    return new BeanPropertyWriter(original) {
+      @Override
+      public void serializeAsField(Object bean, JsonGenerator gen, SerializerProvider prov)
+          throws Exception {
+        gen.writeFieldName(_name);
+        gen.writeString(RedactingSerializer.getRedactedMarker());
+      }
+    };
+  }
+}
+

runtime/service/src/main/java/org/apache/polaris/service/events/json/RedactingSerializer.java

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.polaris.service.events.json;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import java.io.IOException;
+
+/**
+ * A Jackson serializer that redacts string values by replacing them with a redaction marker.
+ *
+ * <p>This serializer is used to prevent sensitive information from being included in serialized
+ * JSON output, particularly for event logging to external systems like CloudWatch.
+ */
+public class RedactingSerializer extends JsonSerializer<Object> {
+  private static final String REDACTED_MARKER = "***REDACTED***";
+
+  @Override
+  public void serialize(Object value, JsonGenerator gen, SerializerProvider serializers)
+      throws IOException {
+    gen.writeString(REDACTED_MARKER);
+  }
+
+  /**
+   * Returns the marker string used to indicate redacted values.
+   *
+   * @return the redaction marker string
+   */
+  public static String getRedactedMarker() {
+    return REDACTED_MARKER;
+  }
+}
+

runtime/service/src/main/java/org/apache/polaris/service/events/json/mixins/IcebergMixins.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.polaris.service.events.json.mixins;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.iceberg.catalog.Namespace;
+
+/** Mixins for Iceberg classes we don't control, to keep JSON concise. */
+public final class IcebergMixins {
+
+  // Private constructor to prevent instantiation
+  private IcebergMixins() {}
+
+  /** Serializes Namespace as an object like: "namespace": ["sales", "north.america"] */
+  public abstract static class NamespaceMixin {
+    @JsonProperty("namespace")
+    public abstract String[] levels();
+  }
+
+  /**
+   * Serializes TableIdentifier as a scalar string like: {"namespace": ["sales", "north.america"],
+   * "name": "transactions"}
+   */
+  public abstract static class TableIdentifierMixin {
+    @JsonProperty("namespace")
+    public abstract Namespace namespace();
+
+    @JsonProperty("name")
+    public abstract String name();
+  }
+}

runtime/service/src/main/java/org/apache/polaris/service/events/json/mixins/PolarisEventBaseMixin.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.polaris.service.events.json.mixins;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.annotation.JsonNaming;
+
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class)
+public abstract class PolarisEventBaseMixin {}