Uploaded image for project: 'Kafka'
  1. Kafka
  2. KAFKA-5891

Cast transformation fails if record schema contains timestamp field

    Details

    • Type: Bug
    • Status: Resolved
    • Priority: Major
    • Resolution: Fixed
    • Affects Version/s: 0.11.0.0
    • Fix Version/s: 1.0.3, 1.1.2, 2.0.1
    • Component/s: KafkaConnect
    • Labels:
      None

      Description

      I have the following simple type cast transformation:

      name=postgresql-source-simple
      connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
      tasks.max=1
      
      connection.url=jdbc:postgresql://localhost:5432/testdb?user=postgres&password=mysecretpassword
      query=SELECT 1::INT as a, '2017-09-14 10:23:54'::TIMESTAMP as b
      
      transforms=Cast
      transforms.Cast.type=org.apache.kafka.connect.transforms.Cast$Value
      transforms.Cast.spec=a:boolean
      
      mode=bulk
      topic.prefix=clients
      

      Which fails with the following exception in runtime:

      [2017-09-14 16:51:01,885] ERROR Task postgresql-source-simple-0 threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:148)
      org.apache.kafka.connect.errors.DataException: Invalid Java object for schema type INT64: class java.sql.Timestamp for field: "null"
      	at org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:239)
      	at org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:209)
      	at org.apache.kafka.connect.data.Struct.put(Struct.java:214)
      	at org.apache.kafka.connect.transforms.Cast.applyWithSchema(Cast.java:152)
      	at org.apache.kafka.connect.transforms.Cast.apply(Cast.java:108)
      	at org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:38)
      	at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:190)
      	at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:168)
      	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:146)
      	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:190)
      	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
      	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
      	at java.lang.Thread.run(Thread.java:745)
      

      If I remove the transforms.* part of the connector it will work correctly. Actually, it doesn't really matter which types I use in the transformation for field 'a', just the existence of a timestamp field brings the exception.

        Attachments

          Issue Links

            Activity

              People

              • Assignee:
                maver1ck Maciej Bryński
                Reporter:
                broartem Artem Plotnikov
              • Votes:
                3 Vote for this issue
                Watchers:
                7 Start watching this issue

                Dates

                • Created:
                  Updated:
                  Resolved: