[runtime] Support per-event-type configurable log levels for event log

api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java

@@ -54,4 +54,32 @@ public class AgentConfigOptions {
     /** The config parameter specifies the unique identifier of job. */
     public static final ConfigOption<String> JOB_IDENTIFIER =
             new ConfigOption<>("job-identifier", String.class, null);
+
+    /**
+     * The global event log level controlling the default verbosity for all event types. Valid
+     * values are "OFF", "STANDARD", and "VERBOSE". Defaults to "STANDARD".
+     */
+    public static final ConfigOption<String> EVENT_LOG_LEVEL =
+            new ConfigOption<>("event-log.level", String.class, "STANDARD");
+
+    /**
+     * The maximum string length for event payloads when logging at STANDARD level. Strings
+     * exceeding this length will be truncated. Defaults to 2000.
+     */
+    public static final ConfigOption<Integer> EVENT_LOG_MAX_STRING_LENGTH =
+            new ConfigOption<>("event-log.standard.max-string-length", Integer.class, 2000);
+
+    /**
+     * The maximum number of array elements to include in event payloads when logging at STANDARD
+     * level. Arrays exceeding this size will be truncated. Defaults to 20.
+     */
+    public static final ConfigOption<Integer> EVENT_LOG_MAX_ARRAY_ELEMENTS =
+            new ConfigOption<>("event-log.standard.max-array-elements", Integer.class, 20);
+
+    /**
+     * The maximum nesting depth for event payloads when logging at STANDARD level. Objects deeper
+     * than this level will be summarized. Defaults to 5.
+     */
+    public static final ConfigOption<Integer> EVENT_LOG_MAX_DEPTH =
+            new ConfigOption<>("event-log.standard.max-depth", Integer.class, 5);
 }

api/src/main/java/org/apache/flink/agents/api/logger/EventLogLevel.java

@@ -0,0 +1,61 @@
+/*
+ * 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.flink.agents.api.logger;
+
+/**
+ * Log level for event logging, controlling the verbosity of event log output.
+ *
+ * <ul>
+ *   <li>{@link #OFF} - No events of this type are logged.
+ *   <li>{@link #STANDARD} - Events are logged with truncated/summarized payloads (default).
+ *   <li>{@link #VERBOSE} - Events are logged with full, untruncated payloads.
+ * </ul>
+ */
+public enum EventLogLevel {
+
+    /** No events of this type are logged. */
+    OFF,
+
+    /** Events are logged with truncated/summarized payloads. This is the default level. */
+    STANDARD,
+
+    /** Events are logged with full, untruncated payloads. */
+    VERBOSE;
+
+    /**
+     * Parses a string value into an {@link EventLogLevel}, case-insensitively.
+     *
+     * @param value the string representation of the log level (e.g., "off", "STANDARD", "Verbose")
+     * @return the corresponding {@link EventLogLevel}
+     * @throws IllegalArgumentException if the value does not match any log level
+     */
+    public static EventLogLevel fromString(String value) {
+        if (value == null) {
+            throw new IllegalArgumentException("EventLogLevel value cannot be null");
+        }
+        try {
+            return valueOf(value.toUpperCase());
+        } catch (IllegalArgumentException e) {
+            throw new IllegalArgumentException(
+                    "Invalid EventLogLevel: '"
+                            + value
+                            + "'. Valid values are: OFF, STANDARD, VERBOSE");
+        }
+    }
+}

api/src/main/java/org/apache/flink/agents/api/logger/EventLoggerConfig.java

@@ -18,8 +18,6 @@
 
 package org.apache.flink.agents.api.logger;
 
-import org.apache.flink.agents.api.EventFilter;
-
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -45,14 +43,11 @@
 public final class EventLoggerConfig {
 
     private final String loggerType;
-    private final EventFilter eventFilter;
     private final Map<String, Object> properties;
 
     /** Private constructor - use {@link #builder()} to create instances. */
-    private EventLoggerConfig(
-            String loggerType, EventFilter eventFilter, Map<String, Object> properties) {
+    private EventLoggerConfig(String loggerType, Map<String, Object> properties) {
         this.loggerType = Objects.requireNonNull(loggerType, "Logger type cannot be null");
-        this.eventFilter = eventFilter == null ? EventFilter.ACCEPT_ALL : eventFilter;
         this.properties = Collections.unmodifiableMap(new HashMap<>(properties));
     }
 
@@ -81,15 +76,6 @@ public String getLoggerType() {
         return loggerType;
     }
 
-    /**
-     * Gets the event filter for this logger configuration.
-     *
-     * @return the EventFilter to apply, never null
-     */
-    public EventFilter getEventFilter() {
-        return eventFilter;
-    }
-
     /**
      * Gets the implementation-specific properties for this logger configuration.
      *
@@ -113,13 +99,12 @@ public boolean equals(Object o) {
         if (o == null || getClass() != o.getClass()) return false;
         EventLoggerConfig that = (EventLoggerConfig) o;
         return Objects.equals(loggerType, that.loggerType)
-                && Objects.equals(eventFilter, that.eventFilter)
                 && Objects.equals(properties, that.properties);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(loggerType, eventFilter, properties);
+        return Objects.hash(loggerType, properties);
     }
 
     @Override
@@ -128,8 +113,6 @@ public String toString() {
                 + "loggerType='"
                 + loggerType
                 + '\''
-                + ", eventFilter="
-                + eventFilter
                 + ", properties="
                 + properties
                 + '}';
@@ -143,7 +126,6 @@ public String toString() {
      */
     public static final class Builder {
         private String loggerType = "file"; // Default to file logger
-        private EventFilter eventFilter = EventFilter.ACCEPT_ALL; // Default to accept all
         private final Map<String, Object> properties = new HashMap<>();
 
         private Builder() {}
@@ -163,18 +145,6 @@ public Builder loggerType(String loggerType) {
             return this;
         }
 
-        /**
-         * Sets the event filter for this configuration.
-         *
-         * @param eventFilter the EventFilter to apply
-         * @return this Builder instance for method chaining
-         * @throws IllegalArgumentException if eventFilter is null
-         */
-        public Builder eventFilter(EventFilter eventFilter) {
-            this.eventFilter = Objects.requireNonNull(eventFilter, "Event filter cannot be null");
-            return this;
-        }
-
         /**
          * Adds a property to the configuration.
          *
@@ -213,7 +183,7 @@ public Builder properties(Map<String, Object> properties) {
          * @return a new EventLoggerConfig instance
          */
         public EventLoggerConfig build() {
-            return new EventLoggerConfig(loggerType, eventFilter, properties);
+            return new EventLoggerConfig(loggerType, properties);
         }
     }
 }

python/flink_agents/api/core_options.py

@@ -91,6 +91,31 @@ class AgentConfigOptions(metaclass=AgentConfigOptionsMeta):
         default=None,
     )
 
+    # Event log level config options
+    EVENT_LOG_LEVEL = ConfigOption(
+        key="event-log.level",
+        config_type=str,
+        default="STANDARD",
+    )
+
+    EVENT_LOG_MAX_STRING_LENGTH = ConfigOption(
+        key="event-log.standard.max-string-length",
+        config_type=int,
+        default=2000,
+    )
+
+    EVENT_LOG_MAX_ARRAY_ELEMENTS = ConfigOption(
+        key="event-log.standard.max-array-elements",
+        config_type=int,
+        default=20,
+    )
+
+    EVENT_LOG_MAX_DEPTH = ConfigOption(
+        key="event-log.standard.max-depth",
+        config_type=int,
+        default=5,
+    )
+
 
 class AgentExecutionOptions:
     """Execution options for Flink Agents."""