Wednesday, December 4, 2019

How to submit REST requests to a distributed Kafka Connect cluster

Goal:

This article shares the examples of curl commands to submit REST requests to a distributed Kafka Connect cluster.

Env:

MapR 6.1 (secured)
mapr-kafka-1.1.1
mapr-kafka-connect-jdbc-4.1.0

Prerequisite:

1. This article will use the same use case documented in this article talking about standalone Kafka Connect. Please understand this use case firstly.
2. To better format the json object, we need to install a tool "jq" on this CentOS 7 env using below command.
yum install epel-release
yum install jq
3. Create a  MapR Streams named "/tmp/hivemeta".
maprcli stream delete -path /tmp/hivemeta
maprcli stream create -path /tmp/hivemeta
4. Put the MySQL JDBC driver to kafka-connect JDBC path. 
cp /opt/mapr/hadoop/hadoop-2.7.0/share/hadoop/common/lib/mysql-connector-java-5.1.25.jar /opt/mapr/kafka-connect-jdbc/kafka-connect-jdbc-4.1.0/share/java/kafka-connect-jdbc/
5. Restart all the Kafka Connect cluster on all nodes.
In this example, this is a 4-nodes Kafka Connect cluster -- v1.poc.com, v2.poc.com, v3.poc.com and v4.poc.com.
maprcli node services -name kafka-connect -action restart -nodes `hostname -f`

Solution:

1. Create a Connector named "mysql-source-dist"

curl -v -X POST https://v1.poc.com:8083/connectors \
--cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr \
-H "Content-Type: application/json" \
--data-binary @- << EOF
{          
    "name": "mysql-source-dist",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
      "tasks.max": "1",
      "connection.url": "jdbc:mysql://v4.poc.com:3306/hive?user=hive&password=hive",
      "table.whitelist": "TBLS",
      "mode": "incrementing",
      "incrementing.column.name": "TBL_ID",
      "topic.prefix": "/tmp/hivemeta:tbls"
 }
}
EOF

2. Check Kafka Connect version

curl https://v1.poc.com:8083 --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr | jq

{
  "version": "1.1.1-mapr-1901",
  "commit": "8185d9b9630a495d",
  "kafka_cluster_id": "7038424823489051793"
}

3. List connector plugins available on this worker

curl https://v1.poc.com:8083/connector-plugins --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr | jq

[
  {
    "class": "io.confluent.connect.hdfs.HdfsSinkConnector",
    "type": "sink",
    "version": "4.1.0"
  },
  {
    "class": "io.confluent.connect.hdfs.tools.SchemaSourceConnector",
    "type": "source",
    "version": "1.1.1-mapr-1901"
  },
  {
    "class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "type": "sink",
    "version": "4.1.0"
  },
  {
    "class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "type": "source",
    "version": "4.1.0"
  },
  {
    "class": "io.confluent.connect.storage.tools.SchemaSourceConnector",
    "type": "source",
    "version": "1.1.1-mapr-1901"
  },
  {
    "class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
    "type": "sink",
    "version": "1.1.1-mapr-1901"
  },
  {
    "class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "type": "source",
    "version": "1.1.1-mapr-1901"
  }
]

4. Listing active connectors on a worker

curl https://v1.poc.com:8083/connectors --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr | jq

[
  "mysql-source-dist"
]

5. Get Connector information 

curl https://v1.poc.com:8083/connectors/mysql-source-dist --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr | jq

{
  "name": "mysql-source-dist",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "mode": "incrementing",
    "incrementing.column.name": "TBL_ID",
    "topic.prefix": "/tmp/hivemeta:tbls",
    "task.user": "mapr",
    "tasks.max": "1",
    "name": "mysql-source-dist",
    "connection.url": "jdbc:mysql://v4.poc.com:3306/hive?user=hive&password=hive",
    "table.whitelist": "TBLS"
  },
  "tasks": [],
  "type": "source"
}

6. Get Connector configuration

curl https://v1.poc.com:8083/connectors/mysql-source-dist/config --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr | jq

{
  "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
  "mode": "incrementing",
  "incrementing.column.name": "TBL_ID",
  "topic.prefix": "/tmp/hivemeta:tbls",
  "task.user": "mapr",
  "tasks.max": "1",
  "name": "mysql-source-dist",
  "connection.url": "jdbc:mysql://v4.poc.com:3306/hive?user=hive&password=hive",
  "table.whitelist": "TBLS"
}

7. Get Connector status 

curl https://v1.poc.com:8083/connectors/mysql-source-dist/status --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr | jq

{
  "name": "mysql-source-dist",
  "connector": {
    "state": "RUNNING",
    "worker_id": "v2.poc.com:8083"
  },
  "tasks": [
    {
      "state": "RUNNING",
      "id": 0,
      "worker_id": "v4.poc.com:8083"
    }
  ],
  "type": "source"
}

8. Pause a Connector 

curl -X PUT https://v1.poc.com:8083/connectors/mysql-source-dist/pause --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr
After a while, if we check the Connector status again, both the Connector and all tasks should be PAUSED.
curl https://v1.poc.com:8083/connectors/mysql-source-dist/status --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr | jq

{
  "name": "mysql-source-dist",
  "connector": {
    "state": "PAUSED",
    "worker_id": "v2.poc.com:8083"
  },
  "tasks": [
    {
      "state": "PAUSED",
      "id": 0,
      "worker_id": "v4.poc.com:8083"
    }
  ],
  "type": "source"
}

9. Resume a Connector

curl -X PUT https://v1.poc.com:8083/connectors/mysql-source-dist/resume --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr
After a while, if we check the Connector status again, both the Connector and all tasks should be RUNNING. 
curl https://v1.poc.com:8083/connectors/mysql-source-dist/status --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr | jq

{
  "name": "mysql-source-dist",
  "connector": {
    "state": "RUNNING",
    "worker_id": "v2.poc.com:8083"
  },
  "tasks": [
    {
      "state": "RUNNING",
      "id": 0,
      "worker_id": "v4.poc.com:8083"
    }
  ],
  "type": "source"
}

10. Updating an exisitng Connector configuration 

curl -v -X PUT https://v1.poc.com:8083/connectors/mysql-source-dist/config \
--cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr \
-H "Content-Type: application/json" \
--data-binary @- << EOF
{
      "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
      "tasks.max": "2",
      "connection.url": "jdbc:mysql://v4.poc.com:3306/hive?user=hive&password=hive",
      "table.whitelist": "TBLS",
      "mode": "incrementing",
      "incrementing.column.name": "TBL_ID",
      "topic.prefix": "/tmp/hivemeta:tbls"
}
EOF

11. Get tasks status 

curl https://v1.poc.com:8083/connectors/mysql-source-dist/tasks --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr | jq

[
  {
    "id": {
      "connector": "mysql-source-dist",
      "task": 0
    },
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
      "mode": "incrementing",
      "incrementing.column.name": "TBL_ID",
      "topic.prefix": "/tmp/hivemeta:tbls",
      "task.user": "mapr",
      "tables": "TBLS",
      "task.class": "io.confluent.connect.jdbc.source.JdbcSourceTask",
      "tasks.max": "2",
      "name": "mysql-source-dist",
      "connection.url": "jdbc:mysql://v4.poc.com:3306/hive?user=hive&password=hive",
      "table.whitelist": "TBLS"
    }
  }
]

12. Get an individual task status

curl https://v1.poc.com:8083/connectors/mysql-source-dist/tasks/0/status --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr | jq

{
  "state": "RUNNING",
  "id": 0,
  "worker_id": "v4.poc.com:8083"
}

13. Restart an individual task

curl -X POST https://v4.poc.com:8083/connectors/mysql-source-dist/tasks/0/restart --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr

14. Delete the Connector 

curl -X DELETE https://v1.poc.com:8083/connectors/mysql-source-dist --cacert /opt/mapr/conf/ssl_truststore.pem  -u mapr:mapr

Reference:

https://mapr.com/docs/61/Kafka/Connect-rest-api.html

Posted by at
Labels: , ,

1 comment:

  1. cpa tax servicesNovember 5, 2024 at 4:38 AM

    Your ability to break down complex ideas into digestible pieces is impressive.

    cpa tax services

    ReplyDelete

Subscribe to: Post Comments (Atom)

Popular Posts