Flink: Upgrade to flink 1.13.2 (apache#3116)

Yanam · Sep 26, 2021 · bae5ce9 · bae5ce9
1 parent 0fda1c8
commit bae5ce9
Show file tree

Hide file tree

Showing 5 changed files with 22 additions and 6 deletions.
diff --git a/api/src/test/java/org/apache/iceberg/AssertHelpers.java b/api/src/test/java/org/apache/iceberg/AssertHelpers.java
@@ -155,4 +155,18 @@ public static void assertEmptyAvroField(GenericRecord record, String field) {
         AvroRuntimeException.class,
         () -> record.get(field));
   }
+
+  /**
+   * Same as {@link AssertHelpers#assertThrowsCause}, but this method compares root cause.
+   */
+  public static void assertThrowsRootCause(String message,
+                                           Class<? extends Exception> expected,
+                                           String containedInMessage,
+                                           Runnable runnable) {
+    Assertions.assertThatThrownBy(runnable::run)
+        .as(message)
+        .getRootCause()
+        .isInstanceOf(expected)
+        .hasMessageContaining(containedInMessage);
+  }
 }
diff --git a/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java b/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java
@@ -28,7 +28,6 @@
 import org.apache.flink.configuration.GlobalConfiguration;
 import org.apache.flink.runtime.util.HadoopUtils;
 import org.apache.flink.table.catalog.Catalog;
-import org.apache.flink.table.descriptors.CatalogDescriptorValidator;
 import org.apache.flink.table.factories.CatalogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
@@ -70,6 +69,9 @@ public class FlinkCatalogFactory implements CatalogFactory {
   public static final String BASE_NAMESPACE = "base-namespace";
   public static final String CACHE_ENABLED = "cache-enabled";
 
+  public static final String TYPE = "type";
+  public static final String PROPERTY_VERSION = "property-version";
+
   /**
    * Create an Iceberg {@link org.apache.iceberg.catalog.Catalog} loader to be used by this Flink catalog adapter.
    *
@@ -104,8 +106,8 @@ static CatalogLoader createCatalogLoader(String name, Map<String, String> proper
   @Override
   public Map<String, String> requiredContext() {
     Map<String, String> context = Maps.newHashMap();
-    context.put(CatalogDescriptorValidator.CATALOG_TYPE, "iceberg");
-    context.put(CatalogDescriptorValidator.CATALOG_PROPERTY_VERSION, "1");
+    context.put(TYPE, "iceberg");
+    context.put(PROPERTY_VERSION, "1");
     return context;
   }
 

diff --git a/flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -13,4 +13,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-org.apache.iceberg.flink.FlinkDynamicTableFactory
+org.apache.iceberg.flink.FlinkDynamicTableFactory
diff --git a/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java b/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
@@ -273,7 +273,7 @@ public void testDowngradeTableToFormatV1ThroughTablePropertyFails() throws Excep
     Assert.assertEquals("should create table using format v2",
         2, ops.refresh().formatVersion());
 
-    AssertHelpers.assertThrowsCause("should fail to downgrade to v1",
+    AssertHelpers.assertThrowsRootCause("should fail to downgrade to v1",
         IllegalArgumentException.class,
         "Cannot downgrade v2 table to v1",
         () -> sql("ALTER TABLE tl SET('format-version'='1')"));

diff --git a/versions.props b/versions.props
@@ -1,7 +1,7 @@
 org.slf4j:* = 1.7.25
 org.apache.avro:avro = 1.10.1
 org.apache.calcite:* = 1.10.0
-org.apache.flink:* = 1.12.1
+org.apache.flink:* = 1.13.2
 org.apache.hadoop:* = 2.7.3
 org.apache.hive:hive-metastore = 2.3.8
 org.apache.hive:hive-serde = 2.3.8