title: How to convert String to Timestamp in kafka connect using transforms and insert into postgres using jdbc sink connector from confluent?
body: <p>Below is my kafka-connect-sink.properties file
  I am using confluent-6.0.1.</p>
  <pre><code>name=enba-sink-postgres
  connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
  connection.url=jdbc:postgresql://IP:PORT/DB
  connection.user=USERNAME
  connection.password=PASSWORD
  tasks.max=1
  topics=postgresInsert
  insert.mode=INSERT
  table.name.format=schema.&quot;tableName&quot;
  auto.create=false
  key.converter.schema.registry.url=http://localhost:8081 
  key.converter.schemas.enable=false
  value.converter.schemas.enable=false
  config.action.reload=restart
  value.converter.schema.registry.url=http://localhost:8081 
  errors.tolerance=all 
  errors.log.enable=true
  errors.log.include.messages=true
  print.key=true 
      
  # Transforms
  transforms=TimestampConverter
  transforms.TimestampConverter.type=org.apache.kafka.connect.transforms.TimestampConverter$Value
  transforms.TimestampConverter.format=yyyy-MM-dd HH:mm:ss
  transforms.TimestampConverter.target.type=Timestamp
  transforms.TimestampConverter.target.field=DATE_TIME
  </code></pre>
  <p>I am using avro data and schema is :</p>
  <pre><code> {\&quot;type\&quot;:\&quot;record\&quot;,\&quot;name\&quot;:\&quot;log\&quot;,\&quot;namespace\&quot;:\&quot;transform.name.space\&quot;,\&quot;fields\&quot;:[{\&quot;name\&quot;:\&quot;TRANSACTION_ID\&quot;,\&quot;type\&quot;:{\&quot;type\&quot;:\&quot;string\&quot;,\&quot;avro.java.string\&quot;:\&quot;String\&quot;},\&quot;avro.java.string\&quot;:\&quot;String\&quot;},{\&quot;name\&quot;:\&quot;MSISDN\&quot;,\&quot;type\&quot;:{\&quot;type\&quot;:\&quot;string\&quot;,\&quot;avro.java.string\&quot;:\&quot;String\&quot;},\&quot;avro.java.string\&quot;:\&quot;String\&quot;},{\&quot;name\&quot;:\&quot;TRIGGER_NAME\&quot;,\&quot;type\&quot;:{\&quot;type\&quot;:\&quot;string\&quot;,\&quot;avro.java.string\&quot;:\&quot;String\&quot;},\&quot;avro.java.string\&quot;:\&quot;String\&quot;},{\&quot;name\&quot;:\&quot;W_ID\&quot;,\&quot;type\&quot;:{\&quot;type\&quot;:\&quot;string\&quot;,\&quot;avro.java.string\&quot;:\&quot;String\&quot;},\&quot;avro.java.string\&quot;:\&quot;String\&quot;},{\&quot;name\&quot;:\&quot;STEP\&quot;,\&quot;type\&quot;:{\&quot;type\&quot;:\&quot;string\&quot;,\&quot;avro.java.string\&quot;:\&quot;String\&quot;},\&quot;avro.java.string\&quot;:\&quot;String\&quot;},{\&quot;name\&quot;:\&quot;REWARD_ID\&quot;,\&quot;type\&quot;:{\&quot;type\&quot;:\&quot;string\&quot;,\&quot;avro.java.string\&quot;:\&quot;String\&quot;},\&quot;avro.java.string\&quot;:\&quot;String\&quot;},{\&quot;name\&quot;:\&quot;CAM_ID\&quot;,\&quot;type\&quot;:{\&quot;type\&quot;:\&quot;string\&quot;,\&quot;avro.java.string\&quot;:\&quot;String\&quot;},\&quot;avro.java.string\&quot;:\&quot;String\&quot;},{\&quot;name\&quot;:\&quot;STATUS\&quot;,\&quot;type\&quot;:{\&quot;type\&quot;:\&quot;string\&quot;,\&quot;avro.java.string\&quot;:\&quot;String\&quot;},\&quot;avro.java.string\&quot;:\&quot;String\&quot;},{\&quot;name\&quot;:\&quot;COMMENTS\&quot;,\&quot;type\&quot;:{\&quot;type\&quot;:\&quot;string\&quot;,\&quot;avro.java.string\&quot;:\&quot;String\&quot;},\&quot;avro.java.string\&quot;:\&quot;String\&quot;},{\&quot;name\&quot;:\&quot;CCR_JSON\&quot;,\&quot;type\&quot;:{\&quot;type\&quot;:\&quot;string\&quot;,\&quot;avro.java.string\&quot;:\&quot;String\&quot;},\&quot;avro.java.string\&quot;:\&quot;String\&quot;},{\&quot;name\&quot;:\&quot;DATE_TIME\&quot;,\&quot;type\&quot;:{\&quot;type\&quot;:\&quot;string\&quot;,\&quot;avro.java.string\&quot;:\&quot;String\&quot;},\&quot;avro.java.string\&quot;:\&quot;String\&quot;}]}
  </code></pre>
  <p>Basically DATE_TIME column in Postgres is of type Timestamp and from avro I tried sending String date and also of type long .
  DATE_TIME = 2022-12-15 14:38:02</p>
  <p>Issue is If I dont use transform then I am getting error :</p>
  <pre><code>ERROR: column &quot;DATE_TIME&quot; is of type timestamp with time zone but expression is of type character varying
  </code></pre>
  <p>And If I use transforms as mentioned above then error is :</p>
  <pre><code>    [2021-02-06 21:47:41,897] ERROR Error encountered in task enba-sink-postgres-0. Executing stage 'TRANSFORMATION' with class 'org.apache.kafka.connect.transforms.TimestampConverter$Value', where consumed record is {topic='enba', partition=0, offset=69, timestamp=1612628261605, timestampType=CreateTime}. (org.apache.kafka.connect.runtime.errors.LogReporter:66)
  org.apache.kafka.connect.errors.ConnectException: Schema Schema{com.package.kafkaconnect.Enbalog:STRUCT} does not correspond to a known timestamp type format
  </code></pre>