Custom connector ignores the status configuration

[petrkarytka]

Hello! I am deploying a custom connector using "confluent_connector" resource. I set status = "PAUSED"

Plan shows

  + config_sensitive    = (sensitive value)
  + id                  = (known after apply)
  + status              = "PAUSED"

Applied successfully

Apply complete! Resources: 1 added, 0 changed, 0 destroyed. Outputs: connectors = { "test-connector" = { "connector_id" = "clcc-**" "connector_status" = "PAUSED" } }

But in fact the connector is in RUNNING state and trying to fetch data from a topic. Is this expected behavior? Could you please confirm the argument is supported for custom connectors as well?

[linouk23]

@petrkarytka thanks for creating this issue!

1. Could you share the output of running terraform plan after applying this plan?
2. Could you confirm that pausing works for your connector when using Confluent Cloud UI?
3. Could you confirm that pausing works for your connector when using the API?

[vemulagopal]

HI, Here is the plan and apply and yes i am able to PAUSE the connector from UI. but i didn't try from the API. Please let me know if you need more information.

# module.kafka_connectors.confluent_connector.gcp_sink["data-ede.edp-event-tracker.subscriptionseats-records-sink"] will be updated in-place
  ~ resource "confluent_connector" "gcp_sink" {
      ~ config_nonsensitive = {
          + "status"                            = "PAUSED"
            # (26 unchanged elements hidden)
        }
        id                  = "lcc-dwg30y"
        # (2 unchanged attributes hidden)

    # (2 unchanged blocks hidden)
}

module.kafka_connectors.confluent_connector.gcp_sink["data-ede.edp-event-tracker.sample-events-sink"]: Modifying... [id=lcc-03n0n9] module.kafka_connectors.confluent_connector.gcp_sink["data-ede.edp-event-tracker.subscriptionseats-records-sink"]: Modifying... [id=lcc-dwg30y] module.kafka_connectors.confluent_connector.gcp_sink["data-ede.edp-event-tracker.subscriptionseats-records-sink"]: Modifications complete after 3s [id=lcc-dwg30y] module.kafka_connectors.confluent_connector.gcp_sink["data-ede.edp-event-tracker.view-records-sink"]: Modifications complete after 4s [id=lcc-zj39y7]

[linouk23]

Thanks!

Could you also try the API directly and see whether it works?

[petrkarytka]

@linouk23

1. Could you share the output of running terraform plan after applying this plan?

  # module.connector["test-connector"].confluent_connector.this will be updated in-place
  ~ resource "confluent_connector" "this" {
        id                  = "clcc-******"
      ~ status              = "RUNNING" -> "PAUSED"
        # (2 unchanged attributes hidden)
        # (2 unchanged blocks hidden)
    }
Plan: 0 to add, 1 to change, 0 to destroy.

module.connector["test-connector"].confluent_connector.this: Modifications complete after 32s [id=clcc-******]
Apply complete! Resources: 0 added, 1 changed, 0 destroyed.
Outputs:
connectors = {
  "test-connector" = {
    "connector_id" = "clcc-******"
    "connector_status" = "PAUSED"
  }
}

output "connector_status" {
  value = resource.confluent_connector.this.status
}

Moreover, it works as expected after the second "apply". The connector updated its status to PAUSED. Then I double checked: destroyed the connector and deployed again without changing anything. It's in RUNNING state.

2. Could you confirm that pausing works for your connector when using Confluent Cloud UI?

I confirm, it works.

3. Could you confirm that pausing works for your connector when using the API?

It works, but throws {"error":null}

[petrkarytka]

@vemulagopal it seems that you put "status" under "config_nonsensitive" so Confluent API treats it as one of nonsensitive parameters.

[linouk23]

it seems that you put "status" under "config_nonsensitive" so Confluent API treats it as one of nonsensitive parameters.

@petrkarytka that's a great catch!

@vemulagopal could you share full TF definition of your resource (excluding sensitive values) so we could reproduce the issue on our side? status attribute should be a top level attribute, that's a great catch!

[linouk23]

@petrkarytka @vemulagopal there's another observation from our team:

Plan shows

  + config_sensitive    = (sensitive value)
  + id                  = (known after apply)
  + status              = "PAUSED"
Applied successfully

Apply complete! Resources: 1 added, 0 changed, 0 destroyed.
Outputs:
connectors = {
"test-connector" = {
"connector_id" = "clcc-******"
"connector_status" = "PAUSED"
}
}

We expect users first not to set status attribute during connector creation and add it separately:

first a user needs to create a connector
resource "confluent_connector" "source" {
  environment {
    id = confluent_environment.staging.id
  }
  kafka_cluster {
    id = confluent_kafka_cluster.basic.id
  }
  ...
}
# terraform apply
and then in order to pause they'd add status = "PAUSED" :
resource "confluent_connector" "source" {
  environment {
    id = confluent_environment.staging.id
  }
  kafka_cluster {
    id = confluent_kafka_cluster.basic.id
  }
  status = "PAUSED"
  ...
}

so you'd see
# Plan: 0 to add, 1 to change, 0 to destroy.

# terraform apply
and then the connector should be paused.

Let us know if that helps!

[petrkarytka]

We expect users first not to set status attribute during connector creation and add it separately:

@linouk23 that makes sense. It wasn't clear whether it's possible to deploy a connector in "PAUSED" status initially.

It would be helpful if you could consider adding support for the initial_state as well. Please see for your reference: https://cwiki.apache.org/confluence/display/KAFKA/KIP-980%3A+Allow+creating+connectors+in+a+stopped+state

[vemulagopal]

@linouk23 , After moving the status property out of config non-sesetive section, it worked fine. Thank you for your help. appreciate it.

[dobeerman]

@linouk23 , After moving the status property out of config non-sesetive section, it worked fine. Thank you for your help. appreciate it.

Most likely you simply applied an upgrade for the state.

When I provision a connector with status = "PAUSED", it initially provisions as RUNNING. The connector is paused only after the second terraform apply.

resource "confluent_connector" "dynamodb_sink" {
  environment {
    id = confluent_environment.demo.id
  }
  kafka_cluster {
    id = confluent_kafka_cluster.demo.id
  }

  # Initial state; Not working.
  status = "PAUSED"

  config_sensitive = {
    "aws.access.key.id"     = jsondecode(data.aws_secretsmanager_secret_version.current.secret_string)["aws_access_key_id"]
    "aws.secret.access.key" = jsondecode(data.aws_secretsmanager_secret_version.current.secret_string)["aws_secret_access_key"]
  }

  config_nonsensitive = {
    "topics"                   = confluent_kafka_topic.topic_orders.topic_name
    "input.data.format"        = "JSON"
    "connector.class"          = "DynamoDbSink"
    "name"                     = "DynamoDbSinkConnector_0"
    "kafka.auth.mode"          = "SERVICE_ACCOUNT"
    "kafka.service.account.id" = confluent_service_account.demo_service_acc.id
    "aws.dynamodb.pk.hash"     = "value.orderid"
    "aws.dynamodb.pk.sort"     = "value.itemid"
    "tasks.max"                = "1"
  }

  lifecycle {
    prevent_destroy = false
  }
}