ConnectException: Exiting WorkerSinkTask due to unrecoverable exception 이슈란?

org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
 at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:609)
 at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:329)
 at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:232)
 at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:201)
 at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:182)
 at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:231)
 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:750)\nCaused by: java.lang.NullPointerException

위 이슈는 Task클래스 내부에서 이슈가 발생하여 태스크가 중단된 상태를 뜻한다. 커넥트 로그를 확인하여 어떤 에러가 발생하였는지 확인하면 된다. 

예를 들어 SinkConnector의 record가 null이 왔을 때 충분히 대응하지 못했을 경우가 있다. try, catch문으로 에러 구문을 충분히 감싸지 않는다면 위 에러와 함께 태스크가 종료(중단)된 상태로 대기할 수 있다.

@Override
public void put(Collection<SinkRecord> records) {
    for (SinkRecord record : records) {
        if (record.value() != null) {