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

org.apache.kafka.connect.transforms.ValueToKey throws NPE

    XMLWordPrintableJSON

    Details

    • Type: Bug
    • Status: Resolved
    • Priority: Minor
    • Resolution: Fixed
    • Affects Version/s: None
    • Fix Version/s: 2.2.3, 2.5.0, 2.3.2, 2.4.1
    • Component/s: KafkaConnect
    • Labels:
      None

      Description

      If a field named in the SMT does not exist a NPE is thrown. This is not helpful to users and should be caught correctly and reported back in a more friendly way.

      For example, importing data from a database with this transform: 

       

      transforms = [ksqlCreateKey, ksqlExtractString]
      transforms.ksqlCreateKey.fields = [ID]
      transforms.ksqlCreateKey.type = class org.apache.kafka.connect.transforms.ValueToKey
      transforms.ksqlExtractString.field = ID
      transforms.ksqlExtractString.type = class org.apache.kafka.connect.transforms.ExtractField$Key
      

      If the field name is id not ID then the task fails : 

      org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
         at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178)
         at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)
         at org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:50)
         at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:293)
         at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:229)
         at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177)
         at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227)
         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:1149)
         at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
         at java.lang.Thread.run(Thread.java:748)
      Caused by: java.lang.NullPointerException
         at org.apache.kafka.connect.transforms.ValueToKey.applyWithSchema(ValueToKey.java:85)
         at org.apache.kafka.connect.transforms.ValueToKey.apply(ValueToKey.java:65)
         at org.apache.kafka.connect.runtime.TransformationChain.lambda$apply$0(TransformationChain.java:50)
         at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
         at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)
         ... 11 more
      
      

       

        Attachments

          Issue Links

            Activity

              People

              • Assignee:
                nigel.liang Nigel Liang
                Reporter:
                rmoff Robin Moffatt
                Reviewer:
                Randall Hauch
              • Votes:
                0 Vote for this issue
                Watchers:
                3 Start watching this issue

                Dates

                • Created:
                  Updated:
                  Resolved: