kafka-jira mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Artem Plotnikov (JIRA)" <j...@apache.org>
Subject [jira] [Updated] (KAFKA-5891) Cast transformation fails if record schema contains timestamp field
Date Thu, 14 Sep 2017 14:18:00 GMT

     [ https://issues.apache.org/jira/browse/KAFKA-5891?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]

Artem Plotnikov updated KAFKA-5891:
-----------------------------------
    Description: 
I have the following simple type cast transformation:
{code}
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
{code}
Which fails with the following exception in runtime:
{code}
[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)
{code}
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.

  was:
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.


> Cast transformation fails if record schema contains timestamp field
> -------------------------------------------------------------------
>
>                 Key: KAFKA-5891
>                 URL: https://issues.apache.org/jira/browse/KAFKA-5891
>             Project: Kafka
>          Issue Type: Bug
>          Components: KafkaConnect
>    Affects Versions: 0.11.0.0
>            Reporter: Artem Plotnikov
>
> I have the following simple type cast transformation:
> {code}
> 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
> {code}
> Which fails with the following exception in runtime:
> {code}
> [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)
> {code}
> 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.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Mime
View raw message