Fix

Author: Felix John

java/Iceberg/S3TableSink/README.md

@@ -47,9 +47,10 @@ The application must have IAM permissions to:
 
 ### Runtime configuration
 
-When running on Amazon Managed Service for Apache Flink the runtime configuration is read from runtime properties.
+When running on Amazon Managed Service for Apache Flink the runtime configuration is read from runtime properties. Make sure that you pass the mandatory parameter `table.bucket.arn`. 
+
 When running locally, the configuration is read from the
-[resources/flink-application-properties-dev.json](./src/main/resources/flink-application-properties-dev.json) file.
+[resources/flink-application-properties-dev.json](./src/main/resources/flink-application-properties-dev.json) file. Make sure that the scope of the dependencides in `pom.xml` is set to `default`when running locally. 
 
 Runtime parameters:
 
@@ -64,19 +65,16 @@ Runtime parameters:
 | `Iceberg` | `upsert.equality.fields` | `symbol`         | Comma separated list of fields used for upsert. It must match partition fields. Required if `operation` = `upsert`. |
 
 ### Checkpoints
-
 Checkpointing must be enabled. Iceberg commits writes on checkpoint. When running locally, the application enables checkpoints programmatically, every 10 seconds.When deployed to Managed Service for Apache Flink, checkpointing is controlled by the application configuration.
 
 
 ### Known limitations
-
 At the moment there are current limitations concerning Flink Iceberg integration with S3 Tables:
 * * Currently, this example needs to be in Flink v1.19, v1.20 isn't supported with the S3 Table Sink yet.
 * Doesn't support Iceberg Table with hidden partitioning
 * Doesn't support adding columns, removing columns, renaming columns or changing columns.
 
 ### Schema and schema evolution
-
 The application must "know" the AVRO schema on start. The schema cannot be dynamically inferred based on the incoming records, for example using a schema registry. This is due to a limitation of the Flink Iceberg integration, that requires knowing the table schema upfront.
 
 This implementation does support schema evolution in the incoming data, as long as new schema versions are FORWARD compatible.
@@ -90,7 +88,6 @@ It is technically possible to fetch the schema on application start from an exte
 schema definition file in an S3 bucket. This is beyond the scope of this example.
 
 ### Running locally, in IntelliJ
-
 You can run this example directly in IntelliJ, without any local Flink cluster or local Flink installation.
 
 See [Running examples locally](https://github.com/nicusX/amazon-managed-service-for-apache-flink-examples/blob/main/java/running-examples-locally.md) for details.

java/Iceberg/S3TableSink/pom.xml

@@ -30,12 +30,15 @@
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-runtime-web</artifactId>
             <version>${flink.version}</version>
+            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-planner_${scala.version}</artifactId>
             <version>${flink.version}</version>
+            <scope>provided</scope>
         </dependency>
+        <!-- Flink Iceberg uses DropWizard metrics -->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-metrics-dropwizard</artifactId>
@@ -45,27 +48,34 @@
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-streaming-java</artifactId>
             <version>${flink.version}</version>
+            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-runtime</artifactId>
             <version>${flink.version}</version>
+            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-connector-datagen</artifactId>
             <version>${flink.version}</version>
+            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-api-java-bridge</artifactId>
             <version>${flink.version}</version>
+            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-common</artifactId>
             <version>${flink.version}</version>
+            <scope>provided</scope>
         </dependency>
+
+        <!-- Flink-Avro Dependencies -->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-avro</artifactId>
@@ -90,6 +100,7 @@
             <groupId>com.amazonaws</groupId>
             <artifactId>aws-kinesisanalytics-runtime</artifactId>
             <version>${kda.runtime.version}</version>
+            <scope>provided</scope>
         </dependency>
 
         <!-- S3 Tables Dependencies -->
@@ -261,13 +272,6 @@
                                     </excludes>
                                 </filter>
                             </filters>
-                            <!-- Shading org.apache.avro, see: https://github.com/apache/iceberg/issues/13481 -->
-                            <relocations>
-                                <relocation>
-                                    <pattern>org.apache.avro</pattern>
-                                    <shadedPattern>shaded.org.apache.avro</shadedPattern>
-                                </relocation>
-                            </relocations>
                             <transformers>
                                 <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                                 <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">

java/Iceberg/S3TableSink/src/main/java/com/amazonaws/services/msf/StreamingJob.java

@@ -80,7 +80,7 @@ public static void main(String[] args) throws Exception {
         }
 
         Map<String, Properties> applicationProperties = loadApplicationProperties(env);
-        icebergProperties = applicationProperties.get("Iceberg");
+        icebergProperties = applicationProperties.getOrDefault("Iceberg", new Properties());
 
         // Get AVRO Schema from the definition bundled with the application
         // Note that the application must "knows" the AVRO schema upfront, i.e. the schema must be either embedded
@@ -107,7 +107,7 @@ public static void main(String[] args) throws Exception {
     }
 
     private static DataStream<GenericRecord> createDataStream(StreamExecutionEnvironment env, Map<String, Properties> applicationProperties, Schema avroSchema) {
-        Properties dataGeneratorProperties = applicationProperties.get("DataGen");
+        Properties dataGeneratorProperties = applicationProperties.getOrDefault("DataGen", new Properties());
         return env.fromSource(
                 createDataGenerator(dataGeneratorProperties, avroSchema),
                 WatermarkStrategy.noWatermarks(),

java/Iceberg/S3TableSink/src/main/java/com/amazonaws/services/msf/iceberg/IcebergSinkBuilder.java

@@ -26,7 +26,7 @@ public class IcebergSinkBuilder {
     private static final String DEFAULT_S3_CATALOG_DB = "default";
     private static final String DEFAULT_ICEBERG_TABLE_NAME = "prices_iceberg";
     private static final String DEFAULT_ICEBERG_PARTITION_FIELDS = "symbol";
-    private static final String DEFAULT_ICEBERG_OPERATION = "upsert";
+    private static final String DEFAULT_ICEBERG_OPERATION = "append";
     private static final String DEFAULT_ICEBERG_UPSERT_FIELDS = "symbol";
 
     /**

java/Iceberg/S3TableSink/src/main/resources/flink-application-properties-dev.json

@@ -9,7 +9,7 @@
     "PropertyGroupId": "Iceberg",
     "PropertyMap": {
 
-      "table.bucket.arn": "your-table-arn",
+      "table.bucket.arn": "<your-table-arn>",
       "catalog.db": "default",
       "catalog.table": "prices_s3table",
       "partition.fields": "symbol",