Flink: Added error handling and default logic for Flink version detection

[gjacoby126]

When FlinkEnvironmentContext initializes, it tries to calculate the exact version of Flink it uses via reflection. This can fail in certain conditions, potentially crashing the Flink pipeline. This patch will catch any failures and give a default based on the known Flink minor version.

Closes #7879

[pvary]

Isn't it a packaging issue, if we have multiple org.apache.flink.streaming.api.datastream.DataStream classes on the classpath?

In this case the load order of these classes might depend on timining, and could cause issues.

[gjacoby126]

@pvary - Yes, it's a packaging issue. In my use case that reproduced the problem, the two instances of DataStream on the classpath are actually identical, due to unrelocated shading. That normally would cause no issues aside from build warnings. Of course we're fixing that too.

Whether the packaging problem causes other bugs or not though, Iceberg's reaction shouldn't be "throw an NPE and crash the Flink pipeline", which is what currently happens and this PR fixes.

[gjacoby126]

Oh, and just fyi, I'm not affiliated with the engineers who posted the original issue this patch fixes; I just came across it while I was investigating. So this has happened at least twice in real-world use.

[pvary]

Why did we get the NullPointerException in this case?
By my experience if we have the class on the classpath multiple times, Java just uses the first one it founds. Was it because the shading messed up with the metadata of the jar?

[gjacoby126]

I don't know for sure, but the shading messing up the metadata is my best guess too.

In addition to the unit test in the patch, I did confirm that the patch resolves the issue in a real Flink cluster.

[pvary]

No need to create a new class for this

[gjacoby126]

I originally made the separate class to make mocking the logic in the unit test easier. Just moved all the logic to FlinkPackage and used static mocking to do the test.

[pvary]

This beats the reason for the automatic version detection. The original goal was to avoid versions in java code. See: #6206

[pvary]

Maybe go for: "Flink-UNKNOWN" in this case?
CC: @stevenzwu, @nastra, @Fokko

[gjacoby126]

I switched the patch to use a single "Flink-UNKNOWN" constant as you suggested.

[pvary]

Why not set this directly in the catch block?

[gjacoby126]

Setting the default in a version == null check also takes care of the case where the reflection returns null instead of an exception, which I saw happen in our TaskManagers.

[pvary]

For me this is more readable:

  public static String version() {
    try {
      String version = getVersionFromJar();
      return version != null ? version : FLINK_UNKNOWN_VERSION;
    } catch (Exception e) {
      return FLINK_UNKNOWN_VERSION;
    }
  }

  @VisibleForTesting
  static String getVersionFromJar() {
    /* Choose {@link DataStream} class because it is one of the core Flink API. */
    return DataStream.class.getPackage().getImplementationVersion();
  }

The reasons:

• Easier to read if the error is handled immediately in the catch block
• Easier to read if the null is handled when we fetch the data
• Add @VisibleForTesting annotation to the method to tell why it is not private

[gjacoby126]

I have made the changes you suggested.

[pvary]

LGTM +1.

Let's see if @stevenzwu, @nastra or @Fokko has any comments

[nastra]

can we please use // instead of /* */ for inline comments?

[nastra]

I think we should store the result in a variable rather than re-fetching the version on every call

[stevenzwu]

+1. this method can be used to initialize the static VERSION variable

[gjacoby126]

@nastra @stevenzwu - Iceberg's checkstyle config forbids naming a non-final static variable an all-caps name; it wants camel-case.

Should I:

1. name it "version" OR
2. Name it VERSION and add a SuppressWarning flag OR
3. Name it VERSION and fix the checkstyle.xml to use the same regex as static constants use

[nastra]

I think we might want to slightly update the implementation to

public class FlinkPackage {

  private static final AtomicReference<String> VERSION = new AtomicReference<>();
  private static final String UNKNOWN_VERSION = "UNKNOWN-VERSION";

  private FlinkPackage() {}

  /** Returns Flink version string like x.y.z */
  public static String version() {
    if (null == VERSION.get()) {
      VERSION.set(initFlinkVersion());
    }

    return VERSION.get();
  }

  private static String initFlinkVersion() {
    try {
      String version = versionFromJar();
      // use unknown version in case exact implementation version can't be found from the jar (this
      // can happen if the DataStream class appears multiple times in the same classpath such as
      // with shading)
      return version != null ? version : UNKNOWN_VERSION;
    } catch (Exception e) {
      return UNKNOWN_VERSION;
    }
  }

  @VisibleForTesting
  static String versionFromJar() {
    // Choose {@link DataStream} class because it is one of the core Flink API
    return DataStream.class.getPackage().getImplementationVersion();
  }
}

Notice that I also removed the get prefix from versionFromJar() as Iceberg typically doesn't use get methods

[pvary]

If we do this, we might not want to create an extra method for initFlinkVersion, just something like this:

  public static String version() {
    if (null == VERSION.get()) {
      try {
        String jarVersion = versionFromJar();
        // use unknown version in case exact implementation version can't be found from the jar (this
        // can happen if the DataStream class appears multiple times in the same classpath such as
        // with shading)
        VERSION.set(jarVersion != null ? jarVersion : UNKNOWN_VERSION);
      } catch (Exception e) {
        VERSION.set(UNKNOWN_VERSION);
      }
    }

    return VERSION.get();
  }

[stevenzwu]

would this simple way work, where initFlinkVersion implements the try-catch?

private static final String VERSION = initFlinkVersion();

[gjacoby126]

@stevenzwu - making VERSION private static final String makes it untestable, because you can't test the normal and default code paths in the same JVM.

@nastra @pvary - Using an AtomicReference would work, and if that's what you want I'll do it, but isn't that more complexity than this logic requires? private static volatile String works just fine, aside from the checkstyle issue mentioned above -- I need to either tweak the checkstyle settings, or give it a camelCase name.

[gjacoby126]

In general, this is a pretty simple patch and it's been going back and forth for a week. Can we come to a consensus on what exactly the patch needs, so that if those things are done it's ready for commit? Thanks!

[gjacoby126]

Uploaded a 1-method version using AtomicReference similar to what @nastra and @pvary were suggesting.

[stevenzwu]

why don't we move the try-catch/null handling to an initVersion method?

private static final String VERSION = initVersion()
...
private static String initVersion() {
     String detectedVersion = null;
      try {
        detectedVersion = versionFromJar();
        // use unknown version in case exact implementation version can't be found from the jar
        // (this can happen if the DataStream class appears multiple times in the same classpath
        // such as with shading)
        detectedVersion = detectedVersion != null ? detectedVersion : FLINK_UNKNOWN_VERSION;
      } catch (Exception e) {
        detectedVersion = FLINK_UNKNOWN_VERSION;
      }
      return detectedVersion;
}

[gjacoby126]

@stevenzwu - the cached value cannot be static final String because that will make the code untestable. (static final AtomicReference<String> is fine, because its internal state is still mutable.) If it's not final, it cannot be named VERSION because the checkstyle config requires non-final static variables to be camelCased.

The current implementation reflects @nastra and @pvary 's last suggestions -- @nastra 's suggestion to use AtomicReference and @pvary 's suggestion to consolidate to one function.

At this point we seem to have contradictory suggestions based on slightly different aesthetic preferences since no two people write code exactly the same. I'm fine implementing any of them, but I can't do all of them.

Is the current implementation ready for commit or can we come to a consensus on what needs to be done to get it ready?

[stevenzwu]

I am not sure it is really necessary to add a unit test for the trivial change. anyway, this is fine to me. I will leave the style decision to @nastra and @pvary

[gjacoby126]

@nastra @pvary , just checking back in. I was wondering if there were any more changes you'd like to see to this patch, or if it's ready for commit?

[nastra]

LGTM, I'll wait a bit in case @pvary has some feedback

[pvary]

Thanks @gjacoby126 for the PR and the persistence until we decided on the right approach!

[gjacoby126]

Thanks for the reviews, @pvary @nastra and @stevenzwu!