Convert long to int64 in Kafka Connect

Question

I am trying to write a custom single message transformation in Kafka Connect. From an int64/Date type field I am producing a long value but I am facing a problem when I am trying to write that long back to an updatedValue with the same schema (int64 type):

    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);

    }

The error is:

[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"

Is there a chance to convert that long value back to an int64 so that it fits the schema

user_5267560 · Answer

I had the same issue. The problem for me was that I specified Date.SCHEMA from kafka.connect.data package as a schema.

With that, you don't need to pass Long value, you can just pass java.util.Date itself. The schema validator will figure it out.

I found this out by digging sources here - 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

The problem here that it find the LOGICAL_TYPE_CLASSES list before SCHEMA_TYPE_CLASSES and therefore trying to see this as Date. Error message there is just misleading.

Convert long to int64 in Kafka Connect

1 Answers