我正在尝试在Kafka Connect中编写自定义的单个消息转换。从一个int64 / Date类型字段中，我正在生成一个长值，但是当我试图用相同的模式（int64类型）将它写回updatedValue时，却遇到了一个问题：
for (Field field : value.schema().fields()) {
final Object origFieldValue = value.get(field);
if (timeField.equals(field.name())){
long date = convertDate(origFieldValue);
updatedValue.put(field, date);
}
updatedValue.put(field, origFieldValue);
}
错误是：
[2020-05-14 14:31:52,120] INFO WorkerSourceTask{id=datechanger} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask)
connect_1 | [2020-05-14 14:31:52,120] INFO WorkerSourceTask{id=datechanger-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask)
connect_1 | [2020-05-14 14:31:52,120] ERROR WorkerSourceTask{id=datechanger-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask)
connect_1 | org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
connect_1 | at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178)
connect_1 | at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)
connect_1 | at org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:50)
connect_1 | at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:320)
connect_1 | at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:245)
connect_1 | at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:184)
connect_1 | at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:234)
connect_1 | at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
connect_1 | at java.util.concurrent.FutureTask.run(FutureTask.java:266)
connect_1 | at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
connect_1 | at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
connect_1 | at java.lang.Thread.run(Thread.java:748)
connect_1 | Caused by: org.apache.kafka.connect.errors.DataException: Invalid Java object for schema type INT64: class java.lang.Long for field: "date"
是否有机会将该long值转换回int64，使其适合模式
我遇到过同样的问题。对我来说，问题是我Date.SCHEMA从kafka.connect.data包中指定了一个架构。
这样，您就不需要传递Long值，只需传递java.util.Date本身即可。模式验证器将解决该问题。
我通过在这里挖掘源代码发现了这一点-https: //github.com/apache/kafka/blob/2.6/connect/api/src/main/java/org/apache/kafka/connect/data/ConnectSchema.java#L69 https://github.com/apache/kafka/blob/2.6/connect/api/src/main/java/org/apache/kafka/connect/data/ConnectSchema.java#L271
这里的问题是，它LOGICAL_TYPE_CLASSES之前找到列表SCHEMA_TYPE_CLASSES，因此尝试将其视为日期。错误消息只是误导。
本文收集自互联网，转载请注明来源。
如有侵权，请联系[email protected] 删除。
我来说两句