Hi Team,
We’re trying to upgrade Flink version to...
# troubleshootingr
Raghunadh Nittala
07/14/2023, 6:43 AMHi Team,
We’re trying to upgrade Flink version to 1.17.1 from 1.6.2. We have developed our Flink pipelines in Kotlin and have koin modules to inject the required dependencies.
Now, we’re facing an AssertionError with calcite metadata with TablePlanner 1.17.1_2.12. We tried excluding calcite from jacoco, but didnt work. Can someone please help on this?
java.lang.AssertionErrorat org.apache.calcite.rel.metadata.MetadataDef.<init>(MetadataDef.java:48)at org.apache.calcite.rel.metadata.MetadataDef.of(MetadataDef.java:64)at org.apache.calcite.rel.metadata.BuiltInMetadata$PercentageOriginalRows.<clinit>(BuiltInMetadata.java:345)at org.apache.calcite.rel.metadata.RelMdPercentageOriginalRows$RelMdPercentageOriginalRowsHandler.getDef(RelMdPercentageOriginalRows.java:231)at org.apache.calcite.rel.metadata.ReflectiveRelMetadataProvider.reflectiveSource(ReflectiveRelMetadataProvider.java:134)at org.apache.calcite.rel.metadata.RelMdPercentageOriginalRows.<clinit>(RelMdPercentageOriginalRows.java:42)at org.apache.calcite.rel.metadata.DefaultRelMetadataProvider.<init>(DefaultRelMetadataProvider.java:42)at org.apache.calcite.rel.metadata.DefaultRelMetadataProvider.<clinit>(DefaultRelMetadataProvider.java:28)at org.apache.calcite.plan.RelOptCluster.<init>(RelOptCluster.java:97)at org.apache.calcite.plan.RelOptCluster.create(RelOptCluster.java:106)at org.apache.flink.table.planner.calcite.FlinkRelOptClusterFactory$.create(FlinkRelOptClusterFactory.scala:36)at org.apache.flink.table.planner.calcite.FlinkRelOptClusterFactory.create(FlinkRelOptClusterFactory.scala)at org.apache.flink.table.planner.delegation.PlannerContext.<init>(PlannerContext.java:132)at org.apache.flink.table.planner.delegation.PlannerBase.<init>(PlannerBase.scala:121)at org.apache.flink.table.planner.delegation.StreamPlanner.<init>(StreamPlanner.scala:65)at org.apache.flink.table.planner.delegation.DefaultPlannerFactory.create(DefaultPlannerFactory.java:65)at org.apache.flink.table.factories.PlannerFactoryUtil.createPlanner(PlannerFactoryUtil.java:58)at org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.create(StreamTableEnvironmentImpl.java:127)at org.apache.flink.table.api.bridge.java.StreamTableEnvironment.create(StreamTableEnvironment.java:122)at org.apache.flink.table.api.bridge.java.StreamTableEnvironment.create(StreamTableEnvironment.java:94)m
Martijn Visser
07/14/2023, 10:10 AMAre you relying on internals/Calcite in your application? If so, I wonder if it’s caused because we have done a Calcite upgrade or changed some of the internals. But there isn’t enough information in this post to understand which interfaces you’re using from Flink and what you’re trying to do
r
Raghunadh Nittala
07/14/2023, 10:16 AMI’m using Flink SQL to consume from an upsert-kafka and converting to a DataStream for further operations. I’m not directly using calcite anywhere, but the StreamTableEnvironment internally is referring to that.
m
Martijn Visser
07/14/2023, 10:50 AMSo are you trying to perform a stateful Flink upgrade? When is the AssertionError thrown?
r
Raghunadh Nittala
07/14/2023, 11:31 AMThe AssertionError is being thrown for a test case. We’re upgrading the version of flink in gradle file and verify the CI builds as first step.
Raghunadh Nittala
07/14/2023, 11:31 AMWe will be doing a stateful upgrade of the operator
m
Martijn Visser
07/14/2023, 11:32 AMYou can’t do that in a minor version with the Table API/SQL
Martijn Visser
07/14/2023, 11:32 AMUnless you’re following FLIP-190
Martijn Visser
07/14/2023, 11:33 AMI think you should check what is the actually error, what’s the code doing there, check the release notes and then determine what might cause it
👀 1
Martijn Visser
07/14/2023, 11:33 AMThere’s not enough info to go by now
r
Raghunadh Nittala
07/14/2023, 11:39 AMAs I mentioned before, in the koin modules we are defining something like :
val flinkEnvModule = module {
single<StreamExecutionEnvironment> {
StreamExecutionEnvironment.getExecutionEnvironment()
}
single {
StreamTableEnvironment.create(get())
}
}
which basically creates the required environments for the pipeline. We have a unit test for the same and thats failing.
should(“<>“) {
_checkKoinModules_(<>) {
withInstance<StreamExecutionEnvironment>()
}
}
After updating flink version this test keeps failing.
m
Martijn Visser
07/14/2023, 11:41 AMAre you indeed upgrading from Flink 1.6 or 1.16?
r
Raghunadh Nittala
07/14/2023, 11:41 AMFlink 1.16.1 to 1.17.1
m
Martijn Visser
07/14/2023, 11:43 AMI don’t know koin modules or Kotlin, so wish I could help you further but I can’t
r
Raghunadh Nittala
07/14/2023, 11:47 AMThats fine Martjin, thanks for your support
s
Sergey Nuyanzin
07/14/2023, 1:56 PMThe problem between JaCoCo and Calcite is that JaCoCo has an agent which adds some methods to Calcite classes on the fly and Calcite does not like that....
If turninng off JaCoCo doesn't help you I'm pretty sure that you probably have other agents e.g. profiler/datadog or something like that which also adds methods to Calcite classes on the fly that Calcite doesn't like
Sergey Nuyanzin
07/14/2023, 2:06 PMif you look at the exception trace you've posted
and compare it against Calcite 1.29.0 source code[1] (1.17.x depends on 1.29.0) you'll see that assertions happens while comparing of length of 2 arrays of methods gathered in some different ways.
I would assume you can debug the code which is failing and set a breakpoint there or just print logs to see difference between these 2 arrays. Since both arrays contains methods by method signature in these array you probably should know the reason. E.g. JaCoCo has a specific jacoco prefix in its method names
[1] https://github.com/apache/calcite/releases/tag/calcite-1.29.0
r
Raghunadh Nittala
07/14/2023, 2:56 PMThanks a lot @Sergey Nuyanzin , will check this
Raghunadh Nittala
07/26/2023, 9:06 AMI solved this by adding the below in my gradle file:
tasks.test {
extensions.configure(JacocoTaskExtension::class) {
excludes = listOf(“org/apache/calcite**“)
}
finalizedBy(tasks.jacocoTestReport) // report is always generated after tests run
}
2 Views