Added error handling and default logic for Flink version detection

flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/util/FlinkPackage.java

@@ -18,16 +18,19 @@
  */
 package org.apache.iceberg.flink.util;
 
-import org.apache.flink.streaming.api.datastream.DataStream;
-
 public class FlinkPackage {
-  /** Choose {@link DataStream} class because it is one of the core Flink API. */
-  private static final String VERSION = DataStream.class.getPackage().getImplementationVersion();
+
+  private static FlinkVersionDetector versionDetector = new FlinkVersionDetector();
 
   private FlinkPackage() {}
 
   /** Returns Flink version string like x.y.z */
   public static String version() {
-    return VERSION;
+    return versionDetector.version();
+  }
+
+  static void setVersionDetector(FlinkVersionDetector detector) {
+    // visible for testing
+    versionDetector = detector;
   }
 }

flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/util/FlinkVersionDetector.java

@@ -0,0 +1,44 @@
+/*
+ * 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.iceberg.flink.util;
+
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+public class FlinkVersionDetector {
+  public String version() {
+    String version = null;
+    try {
+      version = getVersionFromJar();
+    } catch (Exception e) {
+      /* we can't detect the exact implementation version from the jar (this can happen if the DataStream class
+       appears multiple times in the same classpath such as with shading), so the best we can do is say it's
+       1.16.x below
+      */
+    }
+    if (version == null) {
+      version = "1.16.x";
+    }
+    return version;
+  }
+
+  String getVersionFromJar() {
+    /* Choose {@link DataStream} class because it is one of the core Flink API. */
+    return DataStream.class.getPackage().getImplementationVersion();
+  }
+}

flink/v1.16/flink/src/test/java/org/apache/iceberg/flink/util/TestFlinkPackage.java

@@ -18,14 +18,31 @@
  */
 package org.apache.iceberg.flink.util;
 
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 public class TestFlinkPackage {
 
   /** This unit test would need to be adjusted as new Flink version is supported. */
+  @AfterClass
+  public static void cleanup() {
+    FlinkPackage.setVersionDetector(new FlinkVersionDetector());
+  }
+
   @Test
   public void testVersion() {
     Assert.assertEquals("1.16.2", FlinkPackage.version());
   }
+
+  @Test
+  public void testDefaultVersion() {
+    // It's difficult to reproduce a reflection error in a unit test, so we just inject a mocked
+    // fault to test the default logic
+    FlinkVersionDetector detectorWithReflectionError = Mockito.spy(FlinkVersionDetector.class);
+    Mockito.when(detectorWithReflectionError.getVersionFromJar()).thenThrow(RuntimeException.class);
+    FlinkPackage.setVersionDetector(detectorWithReflectionError);
+    Assert.assertEquals("1.16.x", FlinkPackage.version());
+  }
 }

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkPackage.java

@@ -18,16 +18,19 @@
  */
 package org.apache.iceberg.flink.util;
 
-import org.apache.flink.streaming.api.datastream.DataStream;
-
 public class FlinkPackage {
-  /** Choose {@link DataStream} class because it is one of the core Flink API. */
-  private static final String VERSION = DataStream.class.getPackage().getImplementationVersion();
+
+  private static FlinkVersionDetector versionDetector = new FlinkVersionDetector();
 
   private FlinkPackage() {}
 
   /** Returns Flink version string like x.y.z */
   public static String version() {
-    return VERSION;
+    return versionDetector.version();
+  }
+
+  static void setVersionDetector(FlinkVersionDetector detector) {
+    // visible for testing
+    versionDetector = detector;
   }
 }

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkVersionDetector.java

@@ -0,0 +1,44 @@
+/*
+ * 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.iceberg.flink.util;
+
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+public class FlinkVersionDetector {
+  public String version() {
+    String version = null;
+    try {
+      version = getVersionFromJar();
+    } catch (Exception e) {
+      /* we can't detect the exact implementation version from the jar (this can happen if the DataStream class
+       appears multiple times in the same classpath such as with shading), so the best we can do is say it's
+       1.17.x below
+      */
+    }
+    if (version == null) {
+      version = "1.17.x";
+    }
+    return version;
+  }
+
+  String getVersionFromJar() {
+    /* Choose {@link DataStream} class because it is one of the core Flink API. */
+    return DataStream.class.getPackage().getImplementationVersion();
+  }
+}

flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/util/TestFlinkPackage.java

@@ -18,14 +18,31 @@
  */
 package org.apache.iceberg.flink.util;
 
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 public class TestFlinkPackage {
 
   /** This unit test would need to be adjusted as new Flink version is supported. */
+  @AfterClass
+  public static void cleanup() {
+    FlinkPackage.setVersionDetector(new FlinkVersionDetector());
+  }
+
   @Test
   public void testVersion() {
     Assert.assertEquals("1.17.1", FlinkPackage.version());
   }
+
+  @Test
+  public void testDefaultVersion() {
+    // It's difficult to reproduce a reflection error in a unit test, so we just inject a mocked fault to test
+    // the default logic
+    FlinkVersionDetector detectorWithReflectionError = Mockito.spy(FlinkVersionDetector.class);
+    Mockito.when(detectorWithReflectionError.getVersionFromJar()).thenThrow(RuntimeException.class);
+    FlinkPackage.setVersionDetector(detectorWithReflectionError);
+    Assert.assertEquals("1.17.x", FlinkPackage.version());
+  }
 }

flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/util/FlinkPackage.java

@@ -18,16 +18,19 @@
  */
 package org.apache.iceberg.flink.util;
 
-import org.apache.flink.streaming.api.datastream.DataStream;
-
 public class FlinkPackage {
-  /** Choose {@link DataStream} class because it is one of the core Flink API. */
-  private static final String VERSION = DataStream.class.getPackage().getImplementationVersion();
+
+  private static FlinkVersionDetector versionDetector = new FlinkVersionDetector();
 
   private FlinkPackage() {}
 
   /** Returns Flink version string like x.y.z */
   public static String version() {
-    return VERSION;
+    return versionDetector.version();
+  }
+
+  static void setVersionDetector(FlinkVersionDetector detector) {
+    // visible for testing
+    versionDetector = detector;
   }
 }

flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/util/FlinkVersionDetector.java

@@ -0,0 +1,44 @@
+/*
+ * 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.iceberg.flink.util;
+
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+public class FlinkVersionDetector {
+  public String version() {
+    String version = null;
+    try {
+      version = getVersionFromJar();
+    } catch (Exception e) {
+      /* we can't detect the exact implementation version from the jar (this can happen if the DataStream class
+       appears multiple times in the same classpath such as with shading), so the best we can do is say it's
+       1.18.x below
+      */
+    }
+    if (version == null) {
+      version = "1.18.x";
+    }
+    return version;
+  }
+
+  String getVersionFromJar() {
+    /* Choose {@link DataStream} class because it is one of the core Flink API. */
+    return DataStream.class.getPackage().getImplementationVersion();
+  }
+}

flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/util/TestFlinkPackage.java

@@ -18,14 +18,31 @@
  */
 package org.apache.iceberg.flink.util;
 
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 public class TestFlinkPackage {
 
   /** This unit test would need to be adjusted as new Flink version is supported. */
+  @AfterClass
+  public static void cleanup() {
+    FlinkPackage.setVersionDetector(new FlinkVersionDetector());
+  }
+
   @Test
   public void testVersion() {
     Assert.assertEquals("1.18.0", FlinkPackage.version());
   }
+
+  @Test
+  public void testDefaultVersion() {
+    // It's difficult to reproduce a reflection error in a unit test, so we just inject a mocked fault to test
+    // the default logic
+    FlinkVersionDetector detectorWithReflectionError = Mockito.spy(FlinkVersionDetector.class);
+    Mockito.when(detectorWithReflectionError.getVersionFromJar()).thenThrow(RuntimeException.class);
+    FlinkPackage.setVersionDetector(detectorWithReflectionError);
+    Assert.assertEquals("1.18.x", FlinkPackage.version());
+  }
 }