feat: add partitioning, schema validation, catalog properties, and comprehensive tests

- Partition new tables by event_date (day granularity) for efficient
  time-range queries and data lifecycle management
- Validate schema on startup when loading existing tables, failing fast
  if required columns are missing
- Support arbitrary catalog properties via nested <CatalogProperties>
  elements for S3 credentials, REST auth, etc.
- Document catalogImpl short names (hadoop, hive, rest) in Javadoc
- Add IcebergAppenderConfigTest exercising full log4j XML config pipeline
  (plugin discovery, LogManager.getLogger(), context stop flush)
- Add IcebergConcurrencyTest with multi-threaded writer stress tests
- Add IcebergScheduledFlushTest verifying timer-based flush fires
- Add schema validation and partition spec tests to IcebergManagerTest

Author: gsoundar

log4j-iceberg/src/main/java/org/apache/logging/log4j/iceberg/IcebergAppender.java

@@ -16,6 +16,8 @@
  */
 package org.apache.logging.log4j.iceberg;
 
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import org.apache.logging.log4j.core.Appender;
 import org.apache.logging.log4j.core.Core;
@@ -26,24 +28,39 @@
 import org.apache.logging.log4j.core.config.plugins.Plugin;
 import org.apache.logging.log4j.core.config.plugins.PluginBuilderAttribute;
 import org.apache.logging.log4j.core.config.plugins.PluginBuilderFactory;
+import org.apache.logging.log4j.core.config.plugins.PluginElement;
 import org.apache.logging.log4j.core.config.plugins.validation.constraints.Required;
 
 /**
  * Log4j appender that writes log events as rows in an Apache Iceberg table
  * backed by Parquet data files.
  *
+ * <p>The table is partitioned by {@code event_date} (day granularity) for efficient
+ * time-range queries and data lifecycle management.</p>
+ *
+ * <p>The {@code catalogImpl} attribute accepts either a short type name
+ * ({@code "hadoop"}, {@code "hive"}, {@code "rest"}) or a fully qualified class name
+ * (e.g. {@code "org.apache.iceberg.rest.RESTCatalog"}).</p>
+ *
+ * <p>Additional catalog properties (e.g. S3 credentials, REST auth headers) can be
+ * passed via nested {@code <Property>} elements under a {@code <CatalogProperties>}
+ * wrapper.</p>
+ *
  * <p>Configuration example:</p>
  * <pre>
  * &lt;Iceberg name="IcebergAppender"
  *          catalogName="my_catalog"
- *          catalogImpl="org.apache.iceberg.rest.RESTCatalog"
+ *          catalogImpl="rest"
  *          catalogUri="http://localhost:8181"
  *          catalogWarehouse="s3://my-bucket/warehouse"
  *          tableNamespace="logs"
  *          tableName="app_logs"
  *          batchSize="1000"
  *          flushIntervalSeconds="30"&gt;
- *   &lt;PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss} %-5level %logger{36} - %msg%n"/&gt;
+ *   &lt;CatalogProperties&gt;
+ *     &lt;Property name="s3.access-key-id"&gt;AKIA...&lt;/Property&gt;
+ *     &lt;Property name="s3.secret-access-key"&gt;secret&lt;/Property&gt;
+ *   &lt;/CatalogProperties&gt;
  * &lt;/Iceberg&gt;
  * </pre>
  */
@@ -94,7 +111,7 @@ public static class Builder<B extends Builder<B>> extends AbstractAppender.Build
         private String catalogName = "log4j";
 
         @PluginBuilderAttribute
-        @Required(message = "No catalog implementation class provided")
+        @Required(message = "No catalog implementation provided (use 'hadoop', 'hive', 'rest', or a fully qualified class name)")
         private String catalogImpl;
 
         @PluginBuilderAttribute
@@ -116,6 +133,9 @@ public static class Builder<B extends Builder<B>> extends AbstractAppender.Build
         @PluginBuilderAttribute
         private int flushIntervalSeconds = 30;
 
+        @PluginElement("CatalogProperties")
+        private Property[] catalogProperties;
+
         public B setCatalogName(final String catalogName) {
             this.catalogName = catalogName;
             return asBuilder();
@@ -156,8 +176,19 @@ public B setFlushIntervalSeconds(final int flushIntervalSeconds) {
             return asBuilder();
         }
 
+        public B setCatalogProperties(final Property[] catalogProperties) {
+            this.catalogProperties = catalogProperties;
+            return asBuilder();
+        }
+
         @Override
         public IcebergAppender build() {
+            final Map<String, String> extraProperties = new HashMap<>();
+            if (catalogProperties != null) {
+                for (final Property prop : catalogProperties) {
+                    extraProperties.put(prop.getName(), prop.getValue());
+                }
+            }
             final IcebergManager manager = new IcebergManager(
                     getName(),
                     catalogName,
@@ -167,7 +198,8 @@ public IcebergAppender build() {
                     tableNamespace,
                     tableName,
                     batchSize,
-                    flushIntervalSeconds);
+                    flushIntervalSeconds,
+                    extraProperties);
             return new IcebergAppender(getName(), getFilter(), isIgnoreExceptions(), null, manager);
         }
     }

log4j-iceberg/src/main/java/org/apache/logging/log4j/iceberg/IcebergManager.java

@@ -20,9 +20,12 @@
 import java.time.Instant;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
@@ -32,6 +35,7 @@
 import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.DataFile;
+import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.catalog.Catalog;
@@ -65,6 +69,10 @@ class IcebergManager {
             Types.NestedField.optional(6, "thrown", Types.StringType.get()),
             Types.NestedField.required(7, "event_date", Types.DateType.get()));
 
+    static final PartitionSpec PARTITION_SPEC = PartitionSpec.builderFor(LOG_SCHEMA)
+            .day("event_date")
+            .build();
+
     private final String name;
     private final String catalogName;
     private final String catalogImpl;
@@ -74,6 +82,7 @@ class IcebergManager {
     private final String tableName;
     private final int batchSize;
     private final int flushIntervalSeconds;
+    private final Map<String, String> extraCatalogProperties;
 
     private final ReentrantLock lock = new ReentrantLock();
     private List<LogEvent> buffer;
@@ -93,7 +102,8 @@ class IcebergManager {
             final String tableNamespace,
             final String tableName,
             final int batchSize,
-            final int flushIntervalSeconds) {
+            final int flushIntervalSeconds,
+            final Map<String, String> extraCatalogProperties) {
         this.name = name;
         this.catalogName = catalogName;
         this.catalogImpl = catalogImpl;
@@ -103,9 +113,26 @@ class IcebergManager {
         this.tableName = tableName;
         this.batchSize = batchSize;
         this.flushIntervalSeconds = flushIntervalSeconds;
+        this.extraCatalogProperties = extraCatalogProperties != null
+                ? Collections.unmodifiableMap(new HashMap<>(extraCatalogProperties))
+                : Collections.emptyMap();
         this.buffer = new ArrayList<>(batchSize);
     }
 
+    IcebergManager(
+            final String name,
+            final String catalogName,
+            final String catalogImpl,
+            final String catalogUri,
+            final String catalogWarehouse,
+            final String tableNamespace,
+            final String tableName,
+            final int batchSize,
+            final int flushIntervalSeconds) {
+        this(name, catalogName, catalogImpl, catalogUri, catalogWarehouse,
+                tableNamespace, tableName, batchSize, flushIntervalSeconds, null);
+    }
+
     void startup() {
         if (running) {
             return;
@@ -120,6 +147,7 @@ void startup() {
         if (catalogWarehouse != null) {
             catalogProperties.put("warehouse", catalogWarehouse);
         }
+        catalogProperties.putAll(extraCatalogProperties);
 
         catalog = CatalogUtil.buildIcebergCatalog(catalogName, catalogProperties, new Configuration());
         final TableIdentifier tableId = TableIdentifier.of(Namespace.of(tableNamespace), tableName);
@@ -129,10 +157,11 @@ void startup() {
         }
 
         if (!catalog.tableExists(tableId)) {
-            table = catalog.createTable(tableId, LOG_SCHEMA);
-            LOGGER.info("Created Iceberg table {}", tableId);
+            table = catalog.createTable(tableId, LOG_SCHEMA, PARTITION_SPEC);
+            LOGGER.info("Created Iceberg table {} partitioned by event_date", tableId);
         } else {
             table = catalog.loadTable(tableId);
+            validateSchema(table.schema());
             LOGGER.info("Loaded existing Iceberg table {}", tableId);
         }
 
@@ -147,6 +176,25 @@ void startup() {
         running = true;
     }
 
+    void validateSchema(final Schema existingSchema) {
+        final Set<String> required = new HashSet<>();
+        for (final Types.NestedField field : LOG_SCHEMA.columns()) {
+            required.add(field.name());
+        }
+        final Set<String> missing = new HashSet<>();
+        for (final String fieldName : required) {
+            if (existingSchema.findField(fieldName) == null) {
+                missing.add(fieldName);
+            }
+        }
+        if (!missing.isEmpty()) {
+            throw new IllegalStateException(
+                    "Iceberg table " + tableNamespace + "." + tableName
+                            + " is missing required columns: " + missing
+                            + ". Expected schema columns: " + required);
+        }
+    }
+
     void write(final LogEvent event) {
         lock.lock();
         try {
@@ -240,7 +288,7 @@ private DataFile writeParquetFile(final List<LogEvent> events) throws IOExceptio
 
         try (DataWriter<GenericRecord> writer = Parquet.writeData(outputFile)
                 .schema(LOG_SCHEMA)
-                .withSpec(table.spec())
+                .withSpec(PartitionSpec.unpartitioned())
                 .createWriterFunc(GenericParquetWriter::create)
                 .overwrite()
                 .build()) {
@@ -249,7 +297,7 @@ private DataFile writeParquetFile(final List<LogEvent> events) throws IOExceptio
             }
         }
 
-        return org.apache.iceberg.DataFiles.builder(table.spec())
+        return org.apache.iceberg.DataFiles.builder(PartitionSpec.unpartitioned())
                 .withPath(filename)
                 .withFileSizeInBytes(outputFile.toInputFile().getLength())
                 .withRecordCount(events.size())