CODEWITHSUNDEEP
mysqljdbcapache-kafkaapache-kafka-connectdebezium
highfive
highfive

Reputation: 11

Confluent connector tasks have failed

Started the docker containers (cp-kafka-connect-base:7.0.1) and installed the self-managed connector (debezium-connector-mysql:latest). Docker containers works good. When I am trying to configure the connector configuration as below

{
  "name": "MySqlConnectorConnector_0",
  "config": {
    "database.history.consumer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule",
    "database.history.kafka.topic": "pageviews",
    "database.history.consumer.security.protocol": "SASL_SSL",
    "database.history.consumer.ssl.endpoint.identification.algorithm": "https",
    "schema.history.internal.kafka.topic": "PLAIN",
    "database.whitelist": "database-test",
    "database.history.producer.sasl.mechanism": "PLAIN",
    "database.history.producer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule",
    "database.history.producer.ssl.endpoint.identification.algorithm": "https",
    "database.history.producer.security.protocol": "SASL_SSL",
    "database.history.kafka.bootstrap.servers": "localhost:9092",
    "database.server.name": "database-test:3306",
    "schema.history.internal.kafka.bootstrap.servers": "localhost:9092",
    "database.history.consumer.sasl.mechanism": "PLAIN",
    "name": "MySqlConnectorConnector_0",
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",
    "topic.prefix": "test2",
    "database.hostname": "database-test",
    "database.port": "3306",
    "database.user": "admin",
    "database.password": "**********",
    "database.server.id": "11",
    "database.ssl.mode": "disabled",
    "connect.keep.alive": "true",
    "include.schema.changes": "true",
    "inconsistent.schema.handling.mode": "skip"
  }
}

the connector creates but connection task failed

Connector log

    [2023-01-24 17:25:44,827] INFO [Consumer clientId=test3-schemahistory, groupId=test3-schemahistory] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient)
    [2023-01-24 17:25:44,827] WARN [Consumer clientId=test3-schemahistory, groupId=test3-schemahistory] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
    [2023-01-24 17:25:44,827] WARN [Consumer clientId=test3-schemahistory, groupId=test3-schemahistory] Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
    [2023-01-24 17:25:44,831] INFO WorkerSourceTask{id=MySqlConnectorConnector_01-0} Either no records were produced by the task since the last offset commit, or every record has been filtered out by a transformation or dropped due to transformation or conversion errors. (org.apache.kafka.connect.runtime.WorkerSourceTask)
    [2023-01-24 17:25:44,831] WARN Couldn't commit processed log positions with the source database due to a concurrent connector shutdown or restart (io.debezium.connector.common.BaseSourceTask)
    [2023-01-24 17:25:44,873] INFO [Consumer clientId=test3-schemahistory, groupId=test3-schemahistory] Resetting generation due to: consumer pro-actively leaving the group (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
    [2023-01-24 17:25:44,873] INFO [Consumer clientId=test3-schemahistory, groupId=test3-schemahistory] Request joining group due to: consumer pro-actively leaving the group (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
    [2023-01-24 17:25:44,874] INFO Metrics scheduler closed (org.apache.kafka.common.metrics.Metrics)
    [2023-01-24 17:25:44,874] INFO Closing reporter org.apache.kafka.common.metrics.JmxReporter (org.apache.kafka.common.metrics.Metrics)
    [2023-01-24 17:25:44,874] INFO Metrics reporters closed (org.apache.kafka.common.metrics.Metrics)
    [2023-01-24 17:25:44,875] INFO App info kafka.consumer for test3-schemahistory unregistered (org.apache.kafka.common.utils.AppInfoParser)
    [2023-01-24 17:25:44,875] ERROR WorkerSourceTask{id=MySqlConnectorConnector_01-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask)
[2023-01-24 17:33:57,334] INFO [Consumer clientId=test3-schemahistory, groupId=test3-schemahistory] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient)
[2023-01-24 17:33:57,334] WARN [Consumer clientId=test3-schemahistory, groupId=test3-schemahistory] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

Trying to add some extra config parameters, but documentation is not clear for me. Can not feagure out how to fix this and how should connector config looks like.

Upvotes: 0

Views: 572

Answers (1)

OneCricketeer
OneCricketeer

Reputation: 191864

As the error says, schema.history.internal.kafka.bootstrap.servers needs to be a host:port list to Kafka brokers, not a Java class name.

Similarly, schema.history.internal.kafka.topic should be plain string, not necessarily a Java class.

Upvotes: 0

Related Questions