Flink: Add engine name, version to EnvironmentContext (apache#6184)

regadas · Nov 14, 2022 · a9d3630 · a9d3630
1 parent abf3156
commit a9d3630
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 0 deletions.
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
@@ -53,6 +53,7 @@
 import org.apache.flink.util.StringUtils;
 import org.apache.iceberg.CachingCatalog;
 import org.apache.iceberg.DataFile;
+import org.apache.iceberg.EnvironmentContext;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.PartitionField;
 import org.apache.iceberg.PartitionSpec;
@@ -113,6 +114,9 @@ public FlinkCatalog(
     asNamespaceCatalog =
         originalCatalog instanceof SupportsNamespaces ? (SupportsNamespaces) originalCatalog : null;
     closeable = originalCatalog instanceof Closeable ? (Closeable) originalCatalog : null;
+
+    EnvironmentContext.put(EnvironmentContext.ENGINE_NAME, "flink");
+    EnvironmentContext.put(EnvironmentContext.ENGINE_VERSION, "1.14");
   }
 
   @Override

diff --git a/flink/v1.15/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v1.15/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
@@ -53,6 +53,7 @@
 import org.apache.flink.util.StringUtils;
 import org.apache.iceberg.CachingCatalog;
 import org.apache.iceberg.DataFile;
+import org.apache.iceberg.EnvironmentContext;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.PartitionField;
 import org.apache.iceberg.PartitionSpec;
@@ -113,6 +114,9 @@ public FlinkCatalog(
     asNamespaceCatalog =
         originalCatalog instanceof SupportsNamespaces ? (SupportsNamespaces) originalCatalog : null;
     closeable = originalCatalog instanceof Closeable ? (Closeable) originalCatalog : null;
+
+    EnvironmentContext.put(EnvironmentContext.ENGINE_NAME, "flink");
+    EnvironmentContext.put(EnvironmentContext.ENGINE_VERSION, "1.15");
   }
 
   @Override

diff --git a/flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
@@ -53,6 +53,7 @@
 import org.apache.flink.util.StringUtils;
 import org.apache.iceberg.CachingCatalog;
 import org.apache.iceberg.DataFile;
+import org.apache.iceberg.EnvironmentContext;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.PartitionField;
 import org.apache.iceberg.PartitionSpec;
@@ -113,6 +114,9 @@ public FlinkCatalog(
     asNamespaceCatalog =
         originalCatalog instanceof SupportsNamespaces ? (SupportsNamespaces) originalCatalog : null;
     closeable = originalCatalog instanceof Closeable ? (Closeable) originalCatalog : null;
+
+    EnvironmentContext.put(EnvironmentContext.ENGINE_NAME, "flink");
+    EnvironmentContext.put(EnvironmentContext.ENGINE_VERSION, "1.16");
   }
 
   @Override