kafka-connect-transform-common

kafka-connect-transform-common copied to clipboard

I try to add this connector on kafka:

 curl -X POST \
     -H "Content-Type: application/json" \
     --data '
     {"name": "mongo-source-with-ts",
      "config": {
         "connector.class":"com.mongodb.kafka.connect.MongoSourceConnector",
         "connection.uri":"mongodb://localhost:27017",
		 "topic.prefix":"myPrefix",
         "database":"test",
         "collection":"prep",
         "transforms" : "timestampNowField",
         "transforms.timestampNowField.type" : "com.github.jcustenborder.kafka.connect.transform.common.TimestampNowField$Value",
         "transforms.timestampNowField.fields" : "timestamp" 
         }
     }
     ' \
     http://localhost:8083/connectors -w "\n"

I have this error:

org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler\n\tat
 org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:206)\n\tat
 org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:132)\n\tat
 org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:50)\n\tat
 org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:354)\n\tat
 org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:257)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:188)\n\tat
 org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:243)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\n\tat
 java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat
 java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\nCaused by:
 java.lang.UnsupportedOperationException: STRING is not a supported type.\n\tat
 com.github.jcustenborder.kafka.connect.utils.transformation.BaseKeyValueTransformation.processString(BaseKeyValueTransformation.java:56)\n\tat
 com.github.jcustenborder.kafka.connect.utils.transformation.BaseKeyValueTransformation.process(BaseKeyValueTransformation.java:152)\n\tat
 com.github.jcustenborder.kafka.connect.utils.transformation.BaseKeyValueTransformation.apply(BaseKeyValueTransformation.java:192)\n\tat
 org.apache.kafka.connect.runtime.TransformationChain.lambda$apply$0(TransformationChain.java:50)\n\tat
 org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:156)\n\tat
 org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:190)\n\t... 11 more
 11 more

Do you have any idea to resolve this problem?