[
https://issues.apache.org/jira/browse/FLINK-40082?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Sergey Nuyanzin resolved FLINK-40082.
-------------------------------------
Fix Version/s: 2.4.0
Resolution: Fixed
> InvalidProgramException when performing LEFT JOIN with a lookup table that
> has a computed field for Watermark extraction based on metadata field
> (HEADERS)
> ----------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40082
> URL: https://issues.apache.org/jira/browse/FLINK-40082
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
> Affects Versions: 2.2.0, 2.3.0, 2.2.1
> Reporter: Fred Teunissen
> Assignee: Sergey Nuyanzin
> Priority: Major
> Labels: pull-request-available
> Fix For: 2.4.0
>
> Attachments: invalid-program-exception-reproduce-on-docker.tar.gz
>
>
> On Flink 2.2.1 and 2.3.0 the following issue occurs when performing a LEFT
> JOIN with a lookup table that is using a *{*}computed field{*}* to extract
> the Watermark from a metadata field (HEADERS). The Job aborts with an
> InvalidProgramException.
> On Flink 2.1.2 the same job runs without any issues.
> {noformat}
> java.lang.RuntimeException: Could not instantiate generated class
> 'WatermarkGenerator$0'
> at
> org.apache.flink.table.runtime.generated.GeneratedClass.newInstance(GeneratedClass.java:74)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.GeneratedWatermarkGeneratorSupplier.createWatermarkGenerator(GeneratedWatermarkGeneratorSupplier.java:71)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.streaming.api.operators.source.ProgressiveTimestampsAndWatermarks.createMainOutput(ProgressiveTimestampsAndWatermarks.java:128)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.streaming.api.operators.SourceOperator.initializeMainOutput(SourceOperator.java:554)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.streaming.api.operators.SourceOperator.emitNextNotReading(SourceOperator.java:526)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:502)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:646)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:988)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:925)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:973)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:955)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:760)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:569)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at java.base/java.lang.Thread.run(Unknown Source) ~[?:?]
> Caused by: org.apache.flink.util.FlinkRuntimeException:
> org.apache.flink.api.common.InvalidProgramException: Table program cannot be
> compiled. This is a bug. Please file an issue.
> at
> org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:94)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.GeneratedClass.compile(GeneratedClass.java:101)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.GeneratedClass.newInstance(GeneratedClass.java:68)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> ... 16 more
> Caused by:
> org.apache.flink.shaded.guava33.com.google.common.util.concurrent.UncheckedExecutionException:
> org.apache.flink.api.common.InvalidProgramException: Table program cannot be
> compiled. This is a bug. Please file an issue.
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2086)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache.get(LocalCache.java:4017)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4898)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.GeneratedClass.compile(GeneratedClass.java:101)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.GeneratedClass.newInstance(GeneratedClass.java:68)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> ... 16 more
> Caused by: org.apache.flink.api.common.InvalidProgramException: Table program
> cannot be compiled. This is a bug. Please file an issue.
> at
> org.apache.flink.table.runtime.generated.CompileUtils.doCompile(CompileUtils.java:107)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.CompileUtils.lambda$compile$0(CompileUtils.java:92)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4903)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3574)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2316)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2190)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2080)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache.get(LocalCache.java:4017)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4898)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.GeneratedClass.compile(GeneratedClass.java:101)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.GeneratedClass.newInstance(GeneratedClass.java:68)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> ... 16 more
> Caused by: org.codehaus.commons.compiler.CompileException: Line 136, Column
> 86: Thrown exception of type "java.io.UnsupportedEncodingException" is
> neither caught by a "try...catch" block nor declared in the "throws" clause
> of the declaring function
> at org.codehaus.janino.UnitCompiler.compileError(UnitCompiler.java:13080)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler.checkThrownException(UnitCompiler.java:10129)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.invokeConstructor(UnitCompiler.java:8354)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compileGet2(UnitCompiler.java:5640)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.access$10000(UnitCompiler.java:237)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$16.visitNewClassInstance(UnitCompiler.java:4778)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$16.visitNewClassInstance(UnitCompiler.java:4742)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.Java$NewClassInstance.accept(Java.java:5587)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compileGet(UnitCompiler.java:4742)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compileGetValue(UnitCompiler.java:5885)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compileGet2(UnitCompiler.java:5391)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.access$9300(UnitCompiler.java:237)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$16.visitMethodInvocation(UnitCompiler.java:4766)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$16.visitMethodInvocation(UnitCompiler.java:4742)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.Java$MethodInvocation.accept(Java.java:5470)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compileGet(UnitCompiler.java:4742)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compileGetValue(UnitCompiler.java:5885)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:4132)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.access$6300(UnitCompiler.java:237)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$13.visitAssignment(UnitCompiler.java:4088)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$13.visitAssignment(UnitCompiler.java:4071)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.Java$Assignment.accept(Java.java:4864)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:4071)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:2524)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.access$1800(UnitCompiler.java:237)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$6.visitExpressionStatement(UnitCompiler.java:1581)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$6.visitExpressionStatement(UnitCompiler.java:1576)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.Java$ExpressionStatement.accept(Java.java:3209)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:1576)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compileStatements(UnitCompiler.java:1662)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:1647)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.access$1700(UnitCompiler.java:237)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler$6.visitBlock(UnitCompiler.java:1580)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler$6.visitBlock(UnitCompiler.java:1576)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.Java$Block.accept(Java.java:3115)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:1576)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:2650)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.access$1900(UnitCompiler.java:237)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$6.visitIfStatement(UnitCompiler.java:1582)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$6.visitIfStatement(UnitCompiler.java:1576)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.Java$IfStatement.accept(Java.java:3284)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:1576)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compileStatements(UnitCompiler.java:1662)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:3665)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:3330)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler.compileDeclaredMethods(UnitCompiler.java:1448)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler.compileDeclaredMethods(UnitCompiler.java:1421)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:830)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:443)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.access$400(UnitCompiler.java:237)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$3.visitPackageMemberClassDeclaration(UnitCompiler.java:423)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$3.visitPackageMemberClassDeclaration(UnitCompiler.java:419)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.Java$PackageMemberClassDeclaration.accept(Java.java:1688)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:419)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:393)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.access$000(UnitCompiler.java:237)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$2.visitCompilationUnit(UnitCompiler.java:364)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.UnitCompiler$2.visitCompilationUnit(UnitCompiler.java:362)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.Java$CompilationUnit.accept(Java.java:371)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.UnitCompiler.compileUnit(UnitCompiler.java:362)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.SimpleCompiler.cook(SimpleCompiler.java:273)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.codehaus.janino.SimpleCompiler.compileToClassLoader(SimpleCompiler.java:526)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.SimpleCompiler.cook2(SimpleCompiler.java:250)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.SimpleCompiler.cook(SimpleCompiler.java:229)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.janino.SimpleCompiler.cook(SimpleCompiler.java:219)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.commons.compiler.Cookable.cook(Cookable.java:82)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at org.codehaus.commons.compiler.Cookable.cook(Cookable.java:77)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.CompileUtils.doCompile(CompileUtils.java:104)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.CompileUtils.lambda$compile$0(CompileUtils.java:92)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4903)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3574)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2316)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2190)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2080)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache.get(LocalCache.java:4017)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4898)
> ~[flink-dist-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.GeneratedClass.compile(GeneratedClass.java:101)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> at
> org.apache.flink.table.runtime.generated.GeneratedClass.newInstance(GeneratedClass.java:68)
> ~[flink-table-runtime-2.2.1.jar:2.2.1]
> ... 16 more{noformat}
> h2. Reproduce in docker environment.
> To reproduce the issue, you can start the provided Docker Compose environment.
> It will set up a Kafka cluster, Flink cluster and run a Flink SQL job that
> contains the issue.
> Start the Flink SQL job with the following command (from the folder
> containing the `docker-compose.yaml` file):
> {code:bash}
> docker compose run --build --rm sqljob
> {code}
> You can see the error in the logs of the JobManager container (mapped on the
> host to the `./logs/job/` folder).
> The job will fail with an InvalidProgramException.
> To run interactively with the SQL Client, use the following command to start
> the environment without running the job:
> {code:bash}
> docker compose run --build --rm --interactive sqlclient
> {code}
> The `./sql/init.sql` file contains the SQL statements to create the source
> and sink tables and will be automagically executed.
> The `./sql/job.sql` file contains the SQL statements to execute the job that
> performs the LEFT JOIN with the lookup table.
> To cleanup the environment, use the following command:
> {code:bash}
> docker compose down
> {code}
> h2. Steps to Reproduce:
> Create the input source table and output sink table:
> {code:sql}
> CREATE TABLE input (
> id STRING,
> ingestionTime TIMESTAMP(3),
> WATERMARK FOR ingestionTime AS ingestionTime - INTERVAL '0.5' SECOND
> ) WITH (
> 'connector' = 'datagen',
> 'rows-per-second' = '500',
> 'number-of-rows' = '100000'
> );
> CREATE TABLE lookupTbl
> (
> id STRING NOT NULL,
> WATERMARK FOR computedIngestTime AS computedIngestTime - INTERVAL '1.000'
> SECOND,
> headers MAP<STRING,BYTES> NOT NULL METADATA FROM 'headers',
> computedIngestTime AS TO_TIMESTAMP(DECODE(headers['orig_ingest_ts'],
> 'UTF-8')),
> PRIMARY KEY (id) NOT ENFORCED
> ) WITH (
> – connector options
> 'connector' = 'upsert-kafka',
> 'properties.group.id' = 'consumer-group-1',
> 'topic' = 'compacted_topic',
> 'properties.bootstrap.servers' = 'kafka:29092',
> – format options for key options
> 'key.format' = 'raw',
> – format options for value options
> 'value.format' = 'json'
> );
> CREATE TABLE blackhole_table
> (
> id STRING,
> ingest_time TIMESTAMP(3)
> ) WITH (
> 'connector' = 'blackhole'
> );
> {code}
> Execute the job that performs a CROSS JOIN with an inline VALUES clause:
> {code:sql}
> INSERT INTO blackhole_table
> SELECT i.id AS id,
> l.computedIngestTime AS ingest_time
> FROM input AS i
> LEFT JOIN lookupTbl FOR SYSTEM_TIME AS OF i.ingestionTime AS l
> ON i.id = l.id;
> {code}
> h2. Expected Behavior
> Job should run without any errors and produce output to the blackhole sink.
> h2. Actual Behavior:
> Job fails with an InvalidProgramException when trying to compile the
> generated code for the WatermarkGenerator class, due to an uncaught
> UnsupportedEncodingException in the generated code.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
