Kafka Connect Sink 사용할 때 LocalDateTime은 어떻게 하나요??

안녕하세요.
강사님의 강의를 듣고 개인 프로젝트에 적용하고 있습니다.

LocalDateTime이 없을 때는 잘 수행되었는데 제 개인 프로젝트 엔티티에는 LocalDateTime이 있습니다. 이 부분에서 자꾸 에러가 나와서 어떻게 처리해야 하는지 알려주시면 감사하겠습니다.

다음은 제가 LocalDateTime을 처리하기 위해 따로 만든 엔티티 입니다.

package com.dotd.product.kafkaConnectTest;


import jdk.jfr.Name;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.data.annotation.CreatedDate;
import org.springframework.data.jpa.domain.support.AuditingEntityListener;

import javax.persistence.*;
import java.time.LocalDateTime;

@Entity
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
@Table(name = "nametest")
@EntityListeners(AuditingEntityListener.class)
public class NameTest {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Integer id;

    private String name;

    @CreatedDate
    @Column(name = "created_at")
    private LocalDateTime createdAt;


}

또한, Kafka Connect Sink가 한 번 에러가 발생하면 state : FAILED로 바뀌면서 작동을 안 하는 거 같은데 어떻게 정상화 하나요..??

127.0.0.1:8083/connectors/nametest-sink-connect/status

이 요청을 보냈을 때 이렇게 나옵니다.

{
    "name": "nametest-sink-connect",
    "connector": {
        "state": "RUNNING",
        "worker_id": "192.168.0.14:8083"
    },
    "tasks": [
        {
            "id": 0,
            "state": "FAILED",
            "worker_id": "192.168.0.14:8083",
            "trace": "org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.\r\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:611)\r\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:333)\r\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:234)\r\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:203)\r\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:189)\r\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:244)\r\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\r\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\r\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\r\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\r\n\tat java.base/java.lang.Thread.run(Thread.java:834)\r\nCaused by: org.apache.kafka.connect.errors.ConnectException: java.sql.SQLException: Exception chain:\r\njava.sql.BatchUpdateException: Column 'created_at' cannot be null\r\n\r\n\tat io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:124)\r\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:581)\r\n\t... 10 more\r\nCaused by: java.sql.SQLException: Exception chain:\r\njava.sql.BatchUpdateException: Column 'created_at' cannot be null\r\n\r\n\tat io.confluent.connect.jdbc.sink.JdbcSinkTask.getAllMessagesException(JdbcSinkTask.java:154)\r\n\tat io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:104)\r\n\t... 11 more\r\n"
        }
    ],
    "type": "sink"
}

"key.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"key.converter.schema.registry.url": "<스키마 레지스트리 URL>",
"value.converter.schema.registry.url": "<스키마 레지스트리 URL>",
"transforms": "ValueToStruct",
"transforms.ValueToStruct.type": "org.apache.kafka.connect.transforms.HoistField$Value"

{
  "namespace": "com.dotd.product.kafkaConnectTest",
  "type": "record",
  "name": "NameTest",
  "fields": [
    {
      "name": "id",
      "type": "int"
    },
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "createdAt",
      "type": {
        "type": "long",
        "logicalType": "timestamp-millis"
      }
    }
  ]
}