You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Daniel Henneberger <me...@danielhenneberger.com> on 2024/09/06 19:20:08 UTC
Codegen compilation error in docker (only)
I'm running into an issue where my UDF is not being compiled at runtime.
This seems to affect at least flink 1.17 and 1.18. Oddly, this works fine
when I run it in process. It is only in docker where I see the issue.
Anyone have any clues?
```flink.sql
CREATE TEMPORARY FUNCTION IF NOT EXISTS `EpochMilliToTimestamp` AS
'com.datasqrl.time.EpochMilliToTimestamp' LANGUAGE JAVA;
CREATE TEMPORARY TABLE `stu_1` (
`ts` BIGINT NOT NULL,
`src_uuid` STRING NOT NULL,
`ts2` AS `EpochMilliToTimestamp`(`ts`),
WATERMARK FOR `ts2` AS `ts2` - INTERVAL '0.01' SECOND
) WITH (
'format' = 'json',
'properties.bootstrap.servers' = 'kafka:9092',
'properties.group.id' = 'my-group-id',
'topic' = 'my-topic',
'scan.startup.mode' = 'earliest-offset',
'properties.auto.offset.reset' = 'earliest',
'connector' = 'kafka'
);
CREATE TEMPORARY TABLE `stu_2` (
`ts` BIGINT NOT NULL,
`src_uuid` STRING NOT NULL,
`ts2` TIMESTAMP(3) NOT NULL
) WITH (
'connector' = 'blackhole'
);
CREATE VIEW `table$2`
AS
SELECT *
FROM `stu_1`;
EXECUTE STATEMENT SET BEGIN
INSERT INTO `stu_2`
(SELECT *
FROM `table$2`)
;
END;
```
```Udf
package com.datasqrl.time;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.functions.ScalarFunction;
public class EpochMilliToTimestamp extends ScalarFunction {
@DataTypeHint("TIMESTAMP(3)")
public LocalDateTime eval(Long epochMillis) {
return LocalDateTime.ofInstant(Instant.ofEpochMilli(epochMillis),
ZoneOffset.UTC);
}
}
```
deploy-flink-taskmanager-1 | /* 1 */
deploy-flink-taskmanager-1 | /* 2 */ public final class
WatermarkGenerator$0
deploy-flink-taskmanager-1 | /* 3 */ extends
org.apache.flink.table.runtime.generated.WatermarkGenerator {
deploy-flink-taskmanager-1 | /* 4 */
deploy-flink-taskmanager-1 | /* 5 */ private transient
com.datasqrl.time.EpochMilliToTimestamp
function_com$datasqrl$time$EpochMilliToTimestamp;
deploy-flink-taskmanager-1 | /* 6 */ private transient
org.apache.flink.table.data.conversion.TimestampLocalDateTimeConverter
converter$3;
deploy-flink-taskmanager-1 | /* 7 */
deploy-flink-taskmanager-1 | /* 8 */ private transient
org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier.Context
deploy-flink-taskmanager-1 | /* 9 */ context;
deploy-flink-taskmanager-1 | /* 10 */
deploy-flink-taskmanager-1 | /* 11 */
deploy-flink-taskmanager-1 | /* 12 */ public
WatermarkGenerator$0(Object[] references) throws Exception {
deploy-flink-taskmanager-1 | /* 13 */
function_com$datasqrl$time$EpochMilliToTimestamp =
(((com.datasqrl.time.EpochMilliToTimestamp) references[0]));
deploy-flink-taskmanager-1 | /* 14 */ converter$3 =
(((org.apache.flink.table.data.conversion.TimestampLocalDateTimeConverter)
references[1]));
deploy-flink-taskmanager-1 | /* 15 */
deploy-flink-taskmanager-1 | /* 16 */ int len =
references.length;
deploy-flink-taskmanager-1 | /* 17 */ context =
deploy-flink-taskmanager-1 | /* 18 */
(org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier.Context)
references[len-1];
deploy-flink-taskmanager-1 | /* 19 */
deploy-flink-taskmanager-1 | /* 20 */ }
deploy-flink-taskmanager-1 | /* 21 */
deploy-flink-taskmanager-1 | /* 22 */ @Override
deploy-flink-taskmanager-1 | /* 23 */ public void
open(org.apache.flink.configuration.Configuration parameters) throws
Exception {
deploy-flink-taskmanager-1 | /* 24 */
deploy-flink-taskmanager-1 | /* 25 */
function_com$datasqrl$time$EpochMilliToTimestamp.open(new
org.apache.flink.table.planner.codegen.WatermarkGeneratorCodeGeneratorFunctionContextWrapper(context));
deploy-flink-taskmanager-1 | /* 26 */
deploy-flink-taskmanager-1 | /* 27 */
deploy-flink-taskmanager-1 | /* 28 */
converter$3.open(this.getClass().getClassLoader());
deploy-flink-taskmanager-1 | /* 29 */
deploy-flink-taskmanager-1 | /* 30 */ }
deploy-flink-taskmanager-1 | /* 31 */
deploy-flink-taskmanager-1 | /* 32 */ @Override
deploy-flink-taskmanager-1 | /* 33 */ public Long
currentWatermark(org.apache.flink.table.data.RowData row) throws Exception {
deploy-flink-taskmanager-1 | /* 34 */
deploy-flink-taskmanager-1 | /* 35 */ long field$1;
deploy-flink-taskmanager-1 | /* 36 */ boolean isNull$1;
deploy-flink-taskmanager-1 | /* 37 */ java.time.LocalDateTime
externalResult$2;
deploy-flink-taskmanager-1 | /* 38 */
org.apache.flink.table.data.TimestampData result$4;
deploy-flink-taskmanager-1 | /* 39 */ boolean isNull$4;
deploy-flink-taskmanager-1 | /* 40 */ boolean isNull$5;
deploy-flink-taskmanager-1 | /* 41 */
org.apache.flink.table.data.TimestampData result$6;
deploy-flink-taskmanager-1 | /* 42 */ isNull$1 =
row.isNullAt(0);
deploy-flink-taskmanager-1 | /* 43 */ field$1 = -1L;
deploy-flink-taskmanager-1 | /* 44 */ if (!isNull$1) {
deploy-flink-taskmanager-1 | /* 45 */ field$1 =
row.getLong(0);
deploy-flink-taskmanager-1 | /* 46 */ }
deploy-flink-taskmanager-1 | /* 47 */
deploy-flink-taskmanager-1 | /* 48 */
deploy-flink-taskmanager-1 | /* 49 */
deploy-flink-taskmanager-1 | /* 50 */
deploy-flink-taskmanager-1 | /* 51 */ externalResult$2 =
(java.time.LocalDateTime) function_com$datasqrl$time$EpochMilliToTimestamp
deploy-flink-taskmanager-1 | /* 52 */ .eval(isNull$1 ? null :
((java.lang.Long) field$1));
deploy-flink-taskmanager-1 | /* 53 */
deploy-flink-taskmanager-1 | /* 54 */ isNull$4 =
externalResult$2 == null;
deploy-flink-taskmanager-1 | /* 55 */ result$4 = null;
deploy-flink-taskmanager-1 | /* 56 */ if (!isNull$4) {
deploy-flink-taskmanager-1 | /* 57 */ result$4 =
(org.apache.flink.table.data.TimestampData)
converter$3.toInternalOrNull((java.time.LocalDateTime) externalResult$2);
deploy-flink-taskmanager-1 | /* 58 */ }
deploy-flink-taskmanager-1 | /* 59 */
deploy-flink-taskmanager-1 | /* 60 */
deploy-flink-taskmanager-1 | /* 61 */ isNull$5 = isNull$4 ||
false;
deploy-flink-taskmanager-1 | /* 62 */ result$6 = null;
deploy-flink-taskmanager-1 | /* 63 */ if (!isNull$5) {
deploy-flink-taskmanager-1 | /* 64 */
deploy-flink-taskmanager-1 | /* 65 */
deploy-flink-taskmanager-1 | /* 66 */ result$6 =
org.apache.flink.table.data.TimestampData.fromEpochMillis(result$4.getMillisecond()
- ((long) 10L), result$4.getNanoOfMillisecond());
deploy-flink-taskmanager-1 | /* 67 */
deploy-flink-taskmanager-1 | /* 68 */
deploy-flink-taskmanager-1 | /* 69 */ }
deploy-flink-taskmanager-1 | /* 70 */
deploy-flink-taskmanager-1 | /* 71 */ if (isNull$5) {
deploy-flink-taskmanager-1 | /* 72 */ return null;
deploy-flink-taskmanager-1 | /* 73 */ } else {
deploy-flink-taskmanager-1 | /* 74 */ return
result$6.getMillisecond();
deploy-flink-taskmanager-1 | /* 75 */ }
deploy-flink-taskmanager-1 | /* 76 */ }
deploy-flink-taskmanager-1 | /* 77 */
deploy-flink-taskmanager-1 | /* 78 */ @Override
deploy-flink-taskmanager-1 | /* 79 */ public void close() throws
Exception {
deploy-flink-taskmanager-1 | /* 80 */
function_com$datasqrl$time$EpochMilliToTimestamp.close();
deploy-flink-taskmanager-1 | /* 81 */ }
deploy-flink-taskmanager-1 | /* 82 */ }
deploy-flink-taskmanager-1 | /* 83 */
deploy-flink-taskmanager-1 |
deploy-flink-taskmanager-1 | 2024-09-06 19:08:52,126 WARN
org.apache.flink.table.runtime.generated.GeneratedClass [] - Failed
to compile split code, falling back to original code
deploy-flink-taskmanager-1 |
org.apache.flink.util.FlinkRuntimeException:
org.apache.flink.api.common.InvalidProgramException: Table program cannot
be compiled. This is a bug. Please file an issue.
deploy-flink-taskmanager-1 | at
org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:94)
~[flink-table-runtime-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.table.runtime.generated.GeneratedClass.compile(GeneratedClass.java:97)
~[flink-table-runtime-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.table.runtime.generated.GeneratedClass.newInstance(GeneratedClass.java:68)
~[flink-table-runtime-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.table.runtime.generated.GeneratedWatermarkGeneratorSupplier.createWatermarkGenerator(GeneratedWatermarkGeneratorSupplier.java:62)
~[flink-table-runtime-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.streaming.api.operators.source.ProgressiveTimestampsAndWatermarks.createMainOutput(ProgressiveTimestampsAndWatermarks.java:109)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.streaming.api.operators.SourceOperator.initializeMainOutput(SourceOperator.java:460)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.streaming.api.operators.SourceOperator.emitNextNotReading(SourceOperator.java:436)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:412)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:550)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:839)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:788)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952)
[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:931)
[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745)
[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at java.lang.Thread.run(Unknown
Source) [?:?]
deploy-flink-taskmanager-1 | Caused by:
org.apache.flink.shaded.guava30.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.
deploy-flink-taskmanager-1 | at
org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2051)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache.get(LocalCache.java:3962)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4859)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92)
~[flink-table-runtime-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | ... 18 more
deploy-flink-taskmanager-1 | Caused by:
org.apache.flink.api.common.InvalidProgramException: Table program cannot
be compiled. This is a bug. Please file an issue.
deploy-flink-taskmanager-1 | at
org.apache.flink.table.runtime.generated.CompileUtils.doCompile(CompileUtils.java:107)
~[flink-table-runtime-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.table.runtime.generated.CompileUtils.lambda$compile$0(CompileUtils.java:92)
~[flink-table-runtime-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4864)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3529)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2278)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2155)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2045)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache.get(LocalCache.java:3962)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4859)
~[flink-dist-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92)
~[flink-table-runtime-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | ... 18 more
deploy-flink-taskmanager-1 | Caused by:
org.codehaus.commons.compiler.CompileException: Line 25, Column 72: Cannot
determine simple type name "org"
deploy-flink-taskmanager-1 | at
org.codehaus.janino.UnitCompiler.compileError(UnitCompiler.java:12211)
~[flink-table-runtime-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.codehaus.janino.UnitCompiler.getReferenceType(UnitCompiler.java:6833)
~[flink-table-runtime-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.codehaus.janino.UnitCompiler.getReferenceType(UnitCompiler.java:6594)
~[flink-table-runtime-1.17.0.jar:1.17.0]
deploy-flink-taskmanager-1 | at
org.codehaus.janino.UnitCompiler.getReferenceType(UnitCompiler.java:6607)
~[flink-table-runtime-1.17.0.jar:1.17.0]