Flink : Is it necessary to upgrade to flink 1.13 ?

[zhangjun0x01]

Apache Flink 1.13.0 has released, Is it necessary to upgrade to flink 1.13 ?

[zhangjun0x01]

@stevenzwu @openinx

[openinx]

I did few tests to verify whether the iceberg-flink-runtime jar (compiled in flink 1.12 ) could work in flink 1.13, sounds like the only one issue is CatalogDescriptorValidator, it's an Internal API but has been removed in flink 1.13. I think we could remove the usage about Internal API then we could make this jar work in both flink 1.12 and flink 1.13, don't have to upgrade this version to flink 1.13.

[openinx]

Attached the build error message here:

> Task :iceberg-flink:compileJava
/Users/openinx/software/apache-iceberg/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java:31: error: cannot find symbol
import org.apache.flink.table.descriptors.CatalogDescriptorValidator;
                                         ^
  symbol:   class CatalogDescriptorValidator
  location: package org.apache.flink.table.descriptors
/Users/openinx/software/apache-iceberg/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java:105: error: cannot find symbol
    context.put(CatalogDescriptorValidator.CATALOG_TYPE, "iceberg");
                ^
  symbol:   variable CatalogDescriptorValidator
  location: class FlinkCatalogFactory
/Users/openinx/software/apache-iceberg/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java:106: error: cannot find symbol
    context.put(CatalogDescriptorValidator.CATALOG_PROPERTY_VERSION, "1");
                ^
  symbol:   variable CatalogDescriptorValidator
  location: class FlinkCatalogFactory

[zhangjun0x01]

I tested that , when we remove CatalogDescriptorValidator , we can build succeed, but when run the test case, it is failed. I think we may need to do some extra work to be compatible with 1.12 and 1.13

[tiborkiss]

I get stuck with this. I must use Flink 1.12 or 1.13.
I am new to this codebase. I did some research in the flink repository, but I couldn't figure out why it was removed, or to find what is the solution.

The tests fails, because there are tests which assumes this FlinkCatalogFactory based code exists. It cannot be removed without removing the associated tests also.

First, I would have to understand what was the functionality, or reason, why it was used Flink's catalog.

[openinx]

but I couldn't figure out why it was removed, or to find what is the solution.

Because the CatalogDescriptorValidator is a public API which was marked as Internal, the Internal API don't guarantee the compatibility when upgrading from flink 1.12 to flink1.13, so in theory we apache iceberg shouldn't use any of the Internal public API from apache flink unless we have to ( actually we've already followed the rule but still have few depended classes).

I would have to understand what was the functionality, or reason, why it was used Flink's catalog.

Firstly, we apache iceberg need a catalog to manage the iceberg tables, and the catalog also provide the global table lock so that we could implement the iceberg optimistic lock (guarantee the ACID semantic).

Second, it's good to integrate the iceberg catalog implementation to flink catalog interface because all flink sql are built on top of flink catalogs/tables, then we could just use the flink catalog sql to manage the iceberg catalogs and create iceberg tables under flink managed catalog.

[zhangjun0x01]

I did few tests to verify whether the iceberg-flink-runtime jar (compiled in flink 1.12 ) could work in flink 1.13, sounds like the only one issue is CatalogDescriptorValidator, it's an Internal API but has been removed in flink 1.13. I think we could remove the usage about Internal API then we could make this jar work in both flink 1.12 and flink 1.13, don't have to upgrade this version to flink 1.13.

I tested that, if we upgrade to flink 1.13 ，and all UT have passed, some methods of FlinkCatalogFactory need to be modified, because org.apache.flink.table.factories.CatalogFactory interface of flink has changed, the specific code refers to this branch: https://github.com/zhangjun0x01/iceberg/tree/flink1.13 。 I think it is difficult to be compatible with flink 1.12 and 1.13 at the same time

[tiborkiss]

If Ververica 2.5 would came out in short term, which would have flink 1.13, I would not care about the integration with 1.12. Now depends on the the fact how many people are using now 1.11 with iceberg and they would upgrade to skip 1.12 and upgrade to 1.13....

[zhangjun0x01]

If Ververica 2.5 would came out in short term, which would have flink 1.13, I would not care about the integration with 1.12. Now depends on the the fact how many people are using now 1.11 with iceberg and they would upgrade to skip 1.12 and upgrade to 1.13....

During our testing, I found some minor issues that may be flink 1.12, but flink 1.13 fixed these issues, so I upgraded iceberg's flink version to flink 1.13.

[gaborgsomogyi]

+1 on skipping 1.12. @zhangjun0x01 where have you upgraded to 1.13 since I still see 1.12.1 in versions.props on master?

[gaborgsomogyi]

Ah I see it now, testing it...

[silvermissile]

Thanks to this pr , I 'm able to use flink sql 's TO_TIMESTAMP_LTZ() function when insert data into iceberg table . So are Partition transforms on timestamp field.

[kbendick]

Thanks for working on this. I wanted to draw attention to the issue mentioned above (Update FlinkCatalogFactory to implement the new Factory interface). I closed it as basically a duplicate of this one though.

The reason I opened that issue is that several users have reported difficulties in using Iceberg with Flink, specifically in configuring Hadoop.

In some environments, users might not have access to setup the class path properly for including Hadoop. The new Context interface that provides access to the ClassLoader might help with that.

The ultimate goal is to decouple the need for hadoop to be in the environment when it's not actually needed (for example, when using S3FileIO). The thought was that upgrading to 1.13 (and the new FlinkCatalogFactory interface) will be helpful to provide the easier access to the ClassLoader.

In environments where hadoop Configuration isn't used, such as Ververica Platform or AWS Kinesis Data Analytics with S3FileIO, the hope is to remove the need for the Hadoop Configuration entirely.

For Ververica, users have reported a workaround but in Kinesis Data Analytics, it's a bit more difficult (as far as has been reported).

More information on the Hadoop Configurable concern can be found in this issue: #3044

Thanks again @zhangjun0x01!