CODEWITHSUNDEEP
apache-kafkaapache-kafka-connect
Patrick Szalapski
Patrick Szalapski

Reputation: 9439

Kafka Connect with a JdbcConnectionSource connector fails to create task (connector is RUNNING but task is not)

It seems like rather often I create a Kafka Connect connector from the JdbcConnectionSource based on a query, and the connector is created successsfully with status "RUNNING", but no task is created. Looking in the console logs of my container, I see no indication that anything is wrong that I can tell: no errors, no warnings, no explanation of why the task failed. I can get other connectors to work, but sometimes one doesn't.

How can one get more information to troubleshoot when a connector fails to create a RUNNING task?

I'll post an example of my connector config below.

I am using Kafka Connect 5.4.1-ccs.

Connector config (it is an Oracle Database behind JDBC):

{
    "name": "FiscalYear",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "tasks.max": 1,
        "connection.url": "jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=myhost.example.com)(PORT=1521))(LOAD_BALANCE=OFF)(FAILOVER=OFF)(CONNECT_DATA=(SERVER=DEDICATED)(SERVICE_NAME=MY_DB_PRI)(UR=A)))",
        "connection.user":"myuser",
        "connection.password":"mypass",
        "mode": "timestamp",
        "timestamp.column.name": "MAINT_TS",
        "topic.prefix": "MyTeam.MyTopicName",
        "poll.interval.ms": 5000,
        "value.converter" : "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": "false",
        "numeric.mapping": "best_fit",

        "_comment": "The query is wrapped in `select * from ()` so that JdbcSourceConnector can automatically append a WHERE clause.",
        "query": "SELECT * FROM (SELECT fy_nbr, min(fy_strt_dt) fy_strt_dt, max(fy_end_dt) fy_end_dt FROM myuser.fsc_dt fd WHERE fd.fy_nbr >= 2020 and fd.fy_nbr < 2022 group by fy_nbr)/* outer query must have no WHERE clause so that the source connector can append one of its own */"
    }
}

And the Dockerfile that creates my worker:

FROM confluentinc/cp-kafka-connect:latest

# each "CONNECT_" env var refers to a Kafka Connect setting; e.g. CONNECT_REST_PORT refers to setting rest.port
#  see also https://docs.confluent.io/current/connect/references/allconfigs.html

ENV CONNECT_BOOTSTRAP_SERVERS="d.mybroker.example.com:9092"
ENV CONNECT_REST_PORT="8083"
ENV CONNECT_GROUP_ID="MyGroup2" 

ENV CONNECT_CONFIG_STORAGE_TOPIC="MyTeam.ConnectorConfig" 
ENV CONNECT_OFFSET_STORAGE_TOPIC="MyTeam.ConnectorOffsets" 
ENV CONNECT_STATUS_STORAGE_TOPIC="MyTeam.ConnectorStatus" 

ENV CONNECT_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter" 
ENV CONNECT_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" 

ENV CONNECT_INTERNAL_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter"  
ENV CONNECT_INTERNAL_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" 

ENV CONNECT_LOG4J_ROOT_LOGLEVEL="INFO"

COPY ojdbcDrivers /usr/share/java/kafka-connect-jdbc

(I also set the REST advertised hostname environment variable via my Helm chart, so that's why it isn't set in the above.)

After it spins up, I create the connector, then get this from the REST "/status":

{"name":"FiscalYear","connector":{"state":"RUNNING","worker_id":"10.1.2.3:8083"},"tasks":[],"type":"source"}

Upvotes: 5

Views: 2030

Answers (1)

Robin Moffatt
Robin Moffatt

Reputation: 32090

How can one get more information to troubleshoot when a connector fails to create a RUNNING task?

I would increase the logging level on your Kafka Connect worker. Since you're using Apache Kafka 2.4 you can do this dynamically, which is rather useful. Issue this REST API call to your Kafka Connect worker: 

curl -X PUT http://localhost:8083/admin/loggers/io.confluent \
     -H "Content-Type:application/json" -d '{"level": "TRACE"}'

This bumps up all messages for any Confluent connector to TRACE. It also returns a list of the individually loggers, from which you can cherry-pick different loggers and turn their specific loglevel up or down as required. For example: 

curl -X PUT http://localhost:8083/admin/loggers/io.confluent.connect.jdbc.dialect.DatabaseDialects \
     -H "Content-Type:application/json" -d '{"level": "INFO"}'

Ref: https://rmoff.net/2020/01/16/changing-the-logging-level-for-kafka-connect-dynamically/

Upvotes: 2

Related Questions