[
https://issues.apache.org/jira/browse/FLINK-24870?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17442180#comment-17442180
]
wangbaohua edited comment on FLINK-24870 at 11/11/21, 9:18 AM:
---------------------------------------------------------------
https://gitee.com/wang_bh/flink/blob/master/flink-demo/src/main/java/com/asap/demo/rete/ReteDemo4.java
was (Author: JIRAUSER280022):
DataStream<BeanField> kafkaData1 = kafkaData.map(new MapFunction<StandardEvent,
BeanField>() {
@Override
public BeanField map(StandardEvent value) throws Exception {
//构造对象属性
List<JsonField> list = new ArrayList<>();
for (Object jsonField : ruleParse.getTableDefine().getJsonFieldList()) {
JsonField jsonFieldBean = (JsonField) jsonField;
String field = Utils.getUrlByCode(jsonFieldBean.getDbFieldName().toLowerCase());
Class class1 = null;
if (jsonFieldBean.getFieldType().equals("int")) {
class1 = Class.forName("java.lang.Integer");
} else if (jsonFieldBean.getFieldType().equals("string")) {
class1 = Class.forName("java.lang.String");
} else if (jsonFieldBean.getFieldType().equals("long")) {
class1 = Class.forName("java.lang.Long");
} else if (jsonFieldBean.getFieldType().equals("date")) {
class1 = Class.forName("java.util.Date");
}
jsonFieldBean.setDbFieldName(field);
properties.put(field, class1);
list.add(jsonFieldBean);
}
//创建对象
Object stu = Utils.generateObject(properties);
//给实例对象赋值
for (Object jsonField : ruleParse.getTableDefine().getJsonFieldList()) {
JsonField jsonFieldBean = (JsonField) jsonField;
if (jsonFieldBean.getFieldType().equals("date")) {
Utils.setValue(stu, jsonFieldBean.getDbFieldName(),
DateUtil.parseDate(value.getField(jsonFieldBean.getTopicFieldName()),
"yyyy-MM-dd HH:mm:ss:SSS"));
} else if (jsonFieldBean.getFieldType().equals("int")) {
String fieldValueStr = value.getField(jsonFieldBean.getTopicFieldName());
Integer fieldValue = 0;
if (!(fieldValueStr == null || fieldValueStr.isEmpty())) {
fieldValue = Integer.parseInt(fieldValueStr);
}
Utils.setValue(stu, jsonFieldBean.getDbFieldName(), fieldValue);
} else if (jsonFieldBean.getFieldType().equals("long")) {
String fieldValueStr = value.getField(jsonFieldBean.getTopicFieldName());
Long fieldValue = 0L;
if (!(fieldValueStr == null || fieldValueStr.isEmpty())) {
fieldValue = Long.parseLong(fieldValueStr);
}
Utils.setValue(stu, jsonFieldBean.getDbFieldName(), fieldValue);
}
}
return (BeanField)stu;
}
});
kafkaData1.map(new MapFunction<BeanField, String>() {
@Override
public String map(BeanField value) throws Exception {
System.out.println("value:"+value.toString());
return null;
}
});
Table inputTable = blinkStreamTableEnv.fromDataStream(kafkaData1,
Schema.newBuilder()
.column("eventTwoType", "BIGINT")
.column("deviceParentType", "STRING")
.column("type", "STRING")
.column("eventName", "STRING")
.column("directionDesc", "STRING")
.column("srcIp", "STRING")
.column("dstIp", "STRING")
.column("createTime", "TIMESTAMP_LTZ(3)")
.column("snowId", "BIGINT")
.watermark("createTime", "SOURCE_WATERMARK()")
.build());
> Cannot cast "java.util.Date" to "java.time.Instant"
> ---------------------------------------------------
>
> Key: FLINK-24870
> URL: https://issues.apache.org/jira/browse/FLINK-24870
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
> Affects Versions: 1.13.1
> Reporter: wangbaohua
> Priority: Blocker
>
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:582)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:562)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:647)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:537)
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)
> at java.lang.Thread.run(Thread.java:748)
> 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:76)
> at
> org.apache.flink.table.data.conversion.StructuredObjectConverter.open(StructuredObjectConverter.java:80)
> ... 11 more
> Caused by:
> org.apache.flink.shaded.guava18.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.guava18.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2203)
> at
> org.apache.flink.shaded.guava18.com.google.common.cache.LocalCache.get(LocalCache.java:3937)
> at
> org.apache.flink.shaded.guava18.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4739)
> at
> org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:74)
> ... 12 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:89)
> at
> org.apache.flink.table.runtime.generated.CompileUtils.lambda$compile$1(CompileUtils.java:74)
> at
> org.apache.flink.shaded.guava18.com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4742)
> at
> org.apache.flink.shaded.guava18.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3527)
> at
> org.apache.flink.shaded.guava18.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2319)
> at
> org.apache.flink.shaded.guava18.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2282)
> at
> org.apache.flink.shaded.guava18.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2197)
> ... 15 more
> Caused by: org.codehaus.commons.compiler.CompileException: Line 120, Column
> 101: Cannot cast "java.util.Date" to "java.time.Instant"
> at
> org.codehaus.janino.UnitCompiler.compileError(UnitCompiler.java:12211)
> at
> org.codehaus.janino.UnitCompiler.compileGet2(UnitCompiler.java:5051)
> at org.codehaus.janino.UnitCompiler.access$8600(UnitCompiler.java:215)
> at
> org.codehaus.janino.UnitCompiler$16.visitCast(UnitCompiler.java:4418)
> at
> org.codehaus.janino.UnitCompiler$16.visitCast(UnitCompiler.java:4396)
> at org.codehaus.janino.Java$Cast.accept(Java.java:4898)
> at org.codehaus.janino.UnitCompiler.compileGet(UnitCompiler.java:4396)
> at
> org.codehaus.janino.UnitCompiler.compileGet2(UnitCompiler.java:5057)
> at org.codehaus.janino.UnitCompiler.access$8100(UnitCompiler.java:215)
> at
> org.codehaus.janino.UnitCompiler$16$1.visitParenthesizedExpression(UnitCompiler.java:4409)
> at
> org.codehaus.janino.UnitCompiler$16$1.visitParenthesizedExpression(UnitCompiler.java:4400)
> at
> org.codehaus.janino.Java$ParenthesizedExpression.accept(Java.java:4924)
> at
> org.codehaus.janino.UnitCompiler$16.visitLvalue(UnitCompiler.java:4400)
> at
> org.codehaus.janino.UnitCompiler$16.visitLvalue(UnitCompiler.java:4396)
> at org.codehaus.janino.Java$Lvalue.accept(Java.java:4148)
> at org.codehaus.janino.UnitCompiler.compileGet(UnitCompiler.java:4396)
> at
> org.codehaus.janino.UnitCompiler.compileGetValue(UnitCompiler.java:5662)
> at
> org.codehaus.janino.UnitCompiler.compileGet2(UnitCompiler.java:5182)
> at org.codehaus.janino.UnitCompiler.access$9100(UnitCompiler.java:215)
> at
> org.codehaus.janino.UnitCompiler$16.visitMethodInvocation(UnitCompiler.java:4423)
> at
> org.codehaus.janino.UnitCompiler$16.visitMethodInvocation(UnitCompiler.java:4396)
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
