{
            "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
            "connection.url": "jdbc:mysql://10.250.3.157:3306/database?allowPublicKeyRetrieval=true&useSSL=false&autoReconnect=true",
            "connection.username": "user",
            "connection.password": "password",
            "connect.keep.alive": "true",
            "tasks.max": "4",
            "topics.regex": "server.database.*",
            "auto.create": "false",
            "auto.evolve": "true",
            "delete.enabled": "true",
            "insert.mode": "upsert",
            "primary.key.mode": "record_key",
            "errors.tolerance": "all",
            "transforms": "dropTopicPrefix",
            "transforms.dropTopicPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
            "transforms.dropTopicPrefix.regex": "server.database.(.*)",
            "transforms.dropTopicPrefix.replacement": "$1"
    }

  org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:614)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)
	at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:177)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.kafka.connect.errors.ConnectException: JDBC sink connector failure
	at io.debezium.connector.jdbc.JdbcSinkConnectorTask.put(JdbcSinkConnectorTask.java:80)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:583)
	... 11 more
Caused by: org.apache.kafka.connect.errors.ConnectException: Failed to process a sink record
	at io.debezium.connector.jdbc.JdbcChangeEventSink.execute(JdbcChangeEventSink.java:72)
	at io.debezium.connector.jdbc.JdbcSinkConnectorTask.put(JdbcSinkConnectorTask.java:89)
	... 12 more
Caused by: jakarta.persistence.PersistenceException: Converting `org.hibernate.exception.SQLGrammarException` to JPA `PersistenceException` : JDBC exception executing SQL [INSERT INTO user_action_histories (id, created_at, user_id, log, table, tags) VALUES (?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE created_at=VALUES(created_at),user_id=VALUES(user_id),log=VALUES(log),table=VALUES(table),tags=VALUES(tags)]
	at org.hibernate.internal.ExceptionConverterImpl.convert(ExceptionConverterImpl.java:165)
	at org.hibernate.internal.ExceptionConverterImpl.convert(ExceptionConverterImpl.java:175)
	at org.hibernate.query.spi.AbstractQuery.executeUpdate(AbstractQuery.java:654)
	at io.debezium.connector.jdbc.JdbcChangeEventSink.writeUpsert(JdbcChangeEventSink.java:258)
	at io.debezium.connector.jdbc.JdbcChangeEventSink.write(JdbcChangeEventSink.java:217)
	at io.debezium.connector.jdbc.JdbcChangeEventSink.execute(JdbcChangeEventSink.java:69)
	... 13 more
Caused by: org.hibernate.exception.SQLGrammarException: JDBC exception executing SQL [INSERT INTO user_action_histories (id, created_at, user_id, log, table, tags) VALUES (?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE created_at=VALUES(created_at),user_id=VALUES(user_id),log=VALUES(log),table=VALUES(table),tags=VALUES(tags)]
	at org.hibernate.exception.internal.SQLExceptionTypeDelegate.convert(SQLExceptionTypeDelegate.java:64)
	at org.hibernate.exception.internal.StandardSQLExceptionConverter.convert(StandardSQLExceptionConverter.java:56)
	at org.hibernate.engine.jdbc.spi.SqlExceptionHelper.convert(SqlExceptionHelper.java:109)
	at org.hibernate.engine.jdbc.spi.SqlExceptionHelper.convert(SqlExceptionHelper.java:95)
	at org.hibernate.sql.exec.internal.StandardJdbcMutationExecutor.execute(StandardJdbcMutationExecutor.java:97)
	at org.hibernate.query.sql.internal.NativeNonSelectQueryPlanImpl.executeUpdate(NativeNonSelectQueryPlanImpl.java:78)
	at org.hibernate.query.sql.internal.NativeQueryImpl.doExecuteUpdate(NativeQueryImpl.java:820)
	at org.hibernate.query.spi.AbstractQuery.executeUpdate(AbstractQuery.java:643)
	... 16 more
Caused by: java.sql.SQLSyntaxErrorException: You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'table, tags) VALUES ('000168b4-dff9-4d57-a49f-a2fa104b9207', '2022-10-24 11:20:3' at line 1
	at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:121)
	at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:122)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeInternal(ClientPreparedStatement.java:916)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeUpdateInternal(ClientPreparedStatement.java:1061)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeUpdateInternal(ClientPreparedStatement.java:1009)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeLargeUpdate(ClientPreparedStatement.java:1320)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeUpdate(ClientPreparedStatement.java:994)
	at com.mchange.v2.c3p0.impl.NewProxyPreparedStatement.executeUpdate(NewProxyPreparedStatement.java:1502)
	at org.hibernate.sql.exec.internal.StandardJdbcMutationExecutor.execute(StandardJdbcMutationExecutor.java:84)
	... 19 more

| user_action_histories | CREATE TABLE `user_action_histories` (
  `id` char(36) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT 'Идентификатор',
  `created_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Дата',
  `user_id` bigint unsigned DEFAULT NULL COMMENT 'Идентификатор пользователя',
  `log` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT 'Лог',
  `table` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT 'Таблица в которой произошли изменения',
  `tags` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT 'Тэги',
  PRIMARY KEY (`id`),
  KEY `user_action_histories_user_id_foreign` (`user_id`),
  CONSTRAINT `user_action_histories_user_id_foreign` FOREIGN KEY (`user_id`) REFERENCES `users` (`id`) ON DELETE SET NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='История действий пользователей'