Kafka Connect: Automatic deletion of connectors

[viopinto]

Hi,

When any connector is spawned in the Connect Cluster, we observed that it is getting deleted automatically. Could you please advice on this behavior.

curl -XPOST "http://localhost:8083/connectors" -H 'Content-Type: application/json' -d'{
  "name": "elastic-sink-2",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "es-kafkasource",
    "connection.url": "http://elasticsearch-config-es-http:9200",
    "connection.username": "***",
    "connection.password": "***",
    "type.name": "_doc",
    "key.ignore": "true",
    "schema.ignore": "true",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "topic.index.map": "es-kafkasource:es-kafkasink"
  }
}'

Logs:

2021-02-22 06:55:17,526 INFO [Consumer clientId=connector-consumer-elastic-sink-2-0, groupId=connect-elastic-sink-2] Resetting offset for partition es-kafkasource-0 to offset 0. (org.apache.kafka.clients.consumer.internals.SubscriptionState) [task-thread-elastic-sink-2-0]
2021-02-22 06:55:17,526 INFO Index 'es-kafkasink' not found in local cache; checking for existence (io.confluent.connect.elasticsearch.jest.JestElasticsearchClient) [task-thread-elastic-sink-2-0]
2021-02-22 06:55:17,566 INFO Index 'es-kafkasink' exists in Elasticsearch; adding to local cache (io.confluent.connect.elasticsearch.jest.JestElasticsearchClient) [task-thread-elastic-sink-2-0]
2021-02-22 06:55:34,842 INFO Successfully processed removal of connector 'elastic-sink-2' (org.apache.kafka.connect.storage.KafkaConfigBackingStore) [KafkaBasedLog Work Thread - connect-cluster-configs]
2021-02-22 06:55:34,842 INFO [Worker clientId=connect-1, groupId=connect-cluster] Connector elastic-sink-2 config removed (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [KafkaBasedLog Work Thread - connect-cluster-configs]
2021-02-22 06:55:35,488 INFO [Worker clientId=connect-1, groupId=connect-cluster] Handling connector-only config update by stopping connector elastic-sink-2 (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2021-02-22 06:55:35,488 INFO Stopping connector elastic-sink-2 (org.apache.kafka.connect.runtime.Worker) [DistributedHerder-connect-1-1]
2021-02-22 06:55:35,488 INFO Scheduled shutdown for WorkerConnector{id=elastic-sink-2} (org.apache.kafka.connect.runtime.WorkerConnector) [DistributedHerder-connect-1-1]
2021-02-22 06:55:35,489 INFO Completed shutdown for WorkerConnector{id=elastic-sink-2} (org.apache.kafka.connect.runtime.WorkerConnector) [connector-thread-elastic-sink-2]
2021-02-22 06:55:35,490 INFO [Worker clientId=connect-1, groupId=connect-cluster] Rebalance started (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2021-02-22 06:55:35,490 INFO [Worker clientId=connect-1, groupId=connect-cluster] (Re-)joining group (org.apache.kafka.clients.consumer.internals.AbstractCoordinator) [DistributedHerder-connect-1-1]
2021-02-22 06:55:35,569 INFO [Worker clientId=connect-1, groupId=connect-cluster] Successfully joined group with generation 17 (org.apache.kafka.clients.consumer.internals.AbstractCoordinator) [DistributedHerder-connect-1-1]
2021-02-22 06:55:35,571 INFO Stopping connector elastic-sink-2 (org.apache.kafka.connect.runtime.Worker) [StartAndStopExecutor-connect-1-7]
2021-02-22 06:55:35,571 WARN Ignoring stop request for unowned connector elastic-sink-2 (org.apache.kafka.connect.runtime.Worker) [StartAndStopExecutor-connect-1-7]
2021-02-22 06:55:35,571 INFO Stopping task elastic-sink-2-0 (org.apache.kafka.connect.runtime.Worker) [StartAndStopExecutor-connect-1-8]
2021-02-22 06:55:35,571 WARN Ignoring await stop request for non-present connector elastic-sink-2 (org.apache.kafka.connect.runtime.Worker) [StartAndStopExecutor-connect-1-7]
2021-02-22 06:55:35,580 INFO Stopping ElasticsearchSinkTask (io.confluent.connect.elasticsearch.ElasticsearchSinkTask) [task-thread-elastic-sink-2-0]
2021-02-22 06:55:35,624 INFO [Consumer clientId=connector-consumer-elastic-sink-2-0, groupId=connect-elastic-sink-2] Revoke previously assigned partitions es-kafkasource-0 (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator) [task-thread-elastic-sink-2-0]
2021-02-22 06:55:35,624 WARN WorkerSinkTask{id=elastic-sink-2-0} Offset commit failed during close (org.apache.kafka.connect.runtime.WorkerSinkTask) [task-thread-elastic-sink-2-0]
2021-02-22 06:55:35,624 ERROR WorkerSinkTask{id=elastic-sink-2-0} Commit of offsets threw an unexpected exception for sequence number 2: null (org.apache.kafka.connect.runtime.WorkerSinkTask) [task-thread-elastic-sink-2-0]
org.apache.kafka.connect.errors.ConnectException: Stopping
        at io.confluent.connect.elasticsearch.bulk.BulkProcessor.throwIfStopping(BulkProcessor.java:296)
        at io.confluent.connect.elasticsearch.bulk.BulkProcessor.throwIfTerminal(BulkProcessor.java:314)
        at io.confluent.connect.elasticsearch.bulk.BulkProcessor.flush(BulkProcessor.java:378)
        at io.confluent.connect.elasticsearch.ElasticsearchWriter.flush(ElasticsearchWriter.java:343)
        at io.confluent.connect.elasticsearch.ElasticsearchSinkTask.flush(ElasticsearchSinkTask.java:144)
        at org.apache.kafka.connect.sink.SinkTask.preCommit(SinkTask.java:125)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.commitOffsets(WorkerSinkTask.java:386)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.closePartitions(WorkerSinkTask.java:618)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.access$1300(WorkerSinkTask.java:71)
        at org.apache.kafka.connect.runtime.WorkerSinkTask$HandleRebalance.onPartitionsRevoked(WorkerSinkTask.java:694)
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:312)
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:751)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:976)
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:895)
        at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2373)
        at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2340)
        at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2290)
        at org.apache.kafka.common.utils.Utils.closeQuietly(Utils.java:964)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.close(WorkerSinkTask.java:171)
        at org.apache.kafka.connect.runtime.WorkerTask.doClose(WorkerTask.java:164)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:191)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:235)
        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:834)

[scholzj]

Can you share your KafkaConnect custom resource YAML? You should double check you don't have the Connector operator enabled. If you have it enabled, you are expected to manage the connectors using the KafkaConnector custom resources and not using the REST API.

[viopinto]

 apiVersion: kafka.strimzi.io/v1beta1
  kind: KafkaConnect
  metadata:
    name: my-connect-cluster
    annotations:
      strimzi.io/use-connector-resources: "true"
  spec:
    version: 2.6.0
    replicas: 2
    bootstrapServers: my-cluster-kafka-bootstrap:9093
    image: gcr.io/***:0.0.1
    tls:
      trustedCertificates:
        - secretName: my-cluster-cluster-ca-cert
          certificate: ca.crt
    config:
      group.id: connect-cluster
      offset.storage.topic: connect-cluster-offsets
      config.storage.topic: connect-cluster-configs
      status.storage.topic: connect-cluster-status
    resources:
      requests:
        cpu: "1"
        memory: 2Gi
      limits:
        cpu: "2"
        memory: 4Gi

Could you please elaborate if we could make use of both KafkaConnector and REST APIs in parallel?

[scholzj]

Right. So, this part:

    annotations:
      strimzi.io/use-connector-resources: "true"

Enabled the connector operator. So if you want to create, configure or delete your connectors over REST you should remove it.

You cannot really use both REST and the KafkaConnector resources in parallel. When you enable the Connector operator, it will see the KafkaConnector resources as the single source of truth. So if you create a connector directly over REST and it will not have a corresponding KafkaConnector it will be seen as something what should not be there and will be deleted.

What you could do is to have multiple Kafka Connects (https://strimzi.io/docs/operators/latest/full/deploying.html#con-kafka-connect-multiple-instances-str) and use REST on one of them and KafkaConnector on second one.

[viopinto]

Thanks for the inputs @scholzj