fix(clickpipes): add stopped attribute as a remediation to stuck Terraform plan

jkaflik · jkaflik · commit 33ff088c15b7 · 2025-09-22T10:24:50.000+02:00
Current implementation uses `state` attribute in two-way interaction: for status feedback and control on ClickPipe lifecycle by stopping/starting ingestion.

This does not work well in Terraform. Produced state must be equal to plan, meaning, API could return different ClickPipe state, but Terraform would error with unexpected state value for `state` attribute.
This was remediated by introducing an extra condition in `syncClickPipeState` disallowing to persist back to the state if received value was `Failed` or `InternalError`. This was a workaround to not get Terraform complain on provider implementation.
In fact, it was a wrong logic, and it has only hidden a root cause of problem.

This PR splits status feedback and lifecycle control into two separate attributes:
- `state` attribute will be changed to a read-only reporting an actual state back. **This is considered as a breaking change, and will require user to remove `state` from theirs Terraform plan.**
- `stopped` attribute will allow control on pipe state. **If you had your `state = Stopped` previously, you will have to set `stopped = true`.**

This change is necessary to fix a control over pipe's configuration regardless of a current pipe state. `clickhouse_clickpipe` resource is in alpha builds only. It's considered safe to proceed with this change.

Additionally, post-apply status confirmation logic is fixed to retry backoff wait max to 120s.
diff --git a/examples/clickpipe/externally_managed_table/main.tf b/examples/clickpipe/externally_managed_table/main.tf
@@ -46,7 +46,7 @@ resource "clickhouse_clickpipe" "kafka_confluent" {
     replicas = 1
   }
 
-  state = "Running"
+
 
   source = {
     kafka = {
diff --git a/examples/clickpipe/kafka_azure_eventhub/main.tf b/examples/clickpipe/kafka_azure_eventhub/main.tf
@@ -24,7 +24,7 @@ resource "clickhouse_clickpipe" "kafka_azure_eventhub" {
     replicas = 1
   }
 
-  state = "Running"
+
 
   source = {
     kafka = {
diff --git a/examples/clickpipe/kafka_confluent/main.tf b/examples/clickpipe/kafka_confluent/main.tf
@@ -28,8 +28,6 @@ resource "clickhouse_clickpipe" "kafka_confluent" {
     replicas = 1
   }
 
-  state = "Running"
-
   source = {
     kafka = {
       type    = "confluent"
diff --git a/examples/clickpipe/kafka_msk_iam_role/main.tf b/examples/clickpipe/kafka_msk_iam_role/main.tf
@@ -23,7 +23,7 @@ resource "clickhouse_clickpipe" "kafka_msk" {
     replicas = 1
   }
 
-  state = "Running"
+
 
   source = {
     kafka = {
diff --git a/examples/clickpipe/kafka_msk_iam_user/main.tf b/examples/clickpipe/kafka_msk_iam_user/main.tf
@@ -28,7 +28,7 @@ resource "clickhouse_clickpipe" "kafka_msk" {
     replicas = 1
   }
 
-  state = "Running"
+
 
   source = {
     kafka = {
diff --git a/examples/clickpipe/kafka_offset_strategy/main.tf b/examples/clickpipe/kafka_offset_strategy/main.tf
@@ -28,7 +28,7 @@ resource "clickhouse_clickpipe" "kafka_offset_strategy" {
     replicas = 1
   }
 
-  state = "Running"
+
 
   source = {
     kafka = {
diff --git a/examples/clickpipe/kafka_redpanda_scram/main.tf b/examples/clickpipe/kafka_redpanda_scram/main.tf
@@ -28,7 +28,7 @@ resource "clickhouse_clickpipe" "kafka_redpanda" {
     replicas = 1
   }
 
-  state = "Running"
+
 
   source = {
     kafka = {
diff --git a/examples/clickpipe/kafka_schema_registry/main.tf b/examples/clickpipe/kafka_schema_registry/main.tf
@@ -39,7 +39,7 @@ resource "clickhouse_clickpipe" "kafka_schema_registry" {
     replicas = 1
   }
 
-  state = "Running"
+
 
   source = {
     kafka = {
diff --git a/examples/clickpipe/multiple_pipes_example/main.tf b/examples/clickpipe/multiple_pipes_example/main.tf
@@ -32,7 +32,7 @@ resource "clickhouse_clickpipe" "multiple" {
     replica_memory_gb      = 1.0
   }
 
-  state = "Running"
+
 
   source = {
     kafka = {
diff --git a/examples/clickpipe/object_storage_azure_blob/main.tf b/examples/clickpipe/object_storage_azure_blob/main.tf
@@ -23,7 +23,7 @@ variable "azure_path" {
 resource "clickhouse_clickpipe" "azure_blob" {
   name        = "Azure Blob Storage 🚀 ClickPipe"
   service_id = var.service_id
-  state = "Running"
+
   source = {
     object_storage = {
       type    = "azureblobstorage"
diff --git a/examples/clickpipe/object_storage_iam_role/main.tf b/examples/clickpipe/object_storage_iam_role/main.tf
@@ -17,7 +17,7 @@ resource "clickhouse_clickpipe" "kafka_s3" {
 
   service_id = var.service_id
   
-  state = "Running"
+
 
   source = {
     object_storage = {
diff --git a/examples/clickpipe/object_storage_iam_user/main.tf b/examples/clickpipe/object_storage_iam_user/main.tf
@@ -21,7 +21,7 @@ resource "clickhouse_clickpipe" "kafka_s3" {
 
   service_id = var.service_id
   
-  state = "Running"
+
   
   source = {
     object_storage = {
diff --git a/examples/clickpipe/reverse_private_endpoint_msk_pipe/clickpipe.tf b/examples/clickpipe/reverse_private_endpoint_msk_pipe/clickpipe.tf
@@ -12,7 +12,7 @@ resource "clickhouse_clickpipe" "msk" {
     replicas = 1
   }
 
-  state = "Running"
+
 
   source = {
     kafka = {
diff --git a/examples/clickpipe/service_and_clickpipe/main.tf b/examples/clickpipe/service_and_clickpipe/main.tf
@@ -51,7 +51,7 @@ resource "clickhouse_clickpipe" "kafka_confluent" {
     replicas = 1
   }
 
-  state = "Running"
+
 
   source = {
     kafka = {
diff --git a/pkg/internal/api/clickpipe.go b/pkg/internal/api/clickpipe.go
@@ -13,7 +13,9 @@ import (
 
 const (
 	ClickPipeProvisioningState  = "Provisioning"
+	ClickPipeDegradedState      = "Degraded"
 	ClickPipeRunningState       = "Running"
+	ClickPipeStoppingState      = "Stopping"
 	ClickPipeStoppedState       = "Stopped"
 	ClickPipeFailedState        = "Failed"
 	ClickPipeCompletedState     = "Completed"
@@ -237,7 +239,7 @@ func (c *ClientImpl) UpdateClickPipe(ctx context.Context, serviceId string, clic
 	return &clickPipeResponse.Result, nil
 }
 
-func (c *ClientImpl) waitForClickPipe(ctx context.Context, serviceId string, clickPipeId string, stateChecker func(*ClickPipe) bool, maxWaitSeconds uint64) (clickPipe *ClickPipe, err error) {
+func (c *ClientImpl) waitForClickPipe(ctx context.Context, serviceId string, clickPipeId string, stateChecker func(*ClickPipe) bool, maxElapsedTime time.Duration) (clickPipe *ClickPipe, err error) {
 	checkState := func() error {
 		clickPipe, err = c.GetClickPipe(ctx, serviceId, clickPipeId)
 		if err != nil {
@@ -251,18 +253,18 @@ func (c *ClientImpl) waitForClickPipe(ctx context.Context, serviceId string, cli
 		return fmt.Errorf("ClickPipe %s is in state %s", clickPipeId, clickPipe.State)
 	}
 
-	if maxWaitSeconds < 5 {
-		maxWaitSeconds = 5
+	if maxElapsedTime < 5*time.Second {
+		maxElapsedTime = 5
 	}
 
-	err = backoff.Retry(checkState, backoff.WithMaxRetries(backoff.NewConstantBackOff(5*time.Second), maxWaitSeconds/5))
+	err = backoff.Retry(checkState, backoff.NewExponentialBackOff(backoff.WithMaxElapsedTime(maxElapsedTime), backoff.WithMaxInterval(maxElapsedTime/5)))
 	return
 }
 
-func (c *ClientImpl) WaitForClickPipeState(ctx context.Context, serviceId string, clickPipeId string, checker func(string) bool, maxWaitSeconds uint64) (clickPipe *ClickPipe, err error) {
+func (c *ClientImpl) WaitForClickPipeState(ctx context.Context, serviceId string, clickPipeId string, checker func(string) bool, maxWait time.Duration) (clickPipe *ClickPipe, err error) {
 	return c.waitForClickPipe(ctx, serviceId, clickPipeId, func(cp *ClickPipe) bool {
 		return checker(cp.State)
-	}, maxWaitSeconds)
+	}, maxWait)
 }
 
 func (c *ClientImpl) ScalingClickPipe(ctx context.Context, serviceId string, clickPipeId string, request ClickPipeScalingRequest) (*ClickPipe, error) {
diff --git a/pkg/internal/api/client_mock.go b/pkg/internal/api/client_mock.go
diff --git a/pkg/internal/api/interface.go b/pkg/internal/api/interface.go
@@ -2,6 +2,7 @@ package api
 
 import (
 	"context"
+	"time"
 )
 
 type Client interface {
@@ -28,7 +29,7 @@ type Client interface {
 	GetClickPipe(ctx context.Context, serviceId string, clickPipeId string) (*ClickPipe, error)
 	CreateClickPipe(ctx context.Context, serviceId string, clickPipe ClickPipe) (*ClickPipe, error)
 	UpdateClickPipe(ctx context.Context, serviceId string, clickPipeId string, request ClickPipeUpdate) (*ClickPipe, error)
-	WaitForClickPipeState(ctx context.Context, serviceId string, clickPipeId string, stateChecker func(string) bool, maxWaitSeconds uint64) (*ClickPipe, error)
+	WaitForClickPipeState(ctx context.Context, serviceId string, clickPipeId string, stateChecker func(string) bool, maxWait time.Duration) (*ClickPipe, error)
 	ScalingClickPipe(ctx context.Context, serviceId string, clickPipeId string, request ClickPipeScalingRequest) (*ClickPipe, error)
 	ChangeClickPipeState(ctx context.Context, serviceId string, clickPipeId string, command string) (*ClickPipe, error)
 	DeleteClickPipe(ctx context.Context, serviceId string, clickPipeId string) error
diff --git a/pkg/resource/clickpipe.go b/pkg/resource/clickpipe.go
diff --git a/pkg/resource/models/clickpipe_resource.go b/pkg/resource/models/clickpipe_resource.go

Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,7 @@ resource "clickhouse_clickpipe" "kafka_confluent" {`
`46`	`46`	`replicas = 1`
`47`	`47`	`}`
`48`	`48`
`49`		`- state = "Running"`
	`49`	`+`
`50`	`50`
`51`	`51`	`source = {`
`52`	`52`	`kafka = {`
Original file line number	Diff line number	Diff line change
`@@ -24,7 +24,7 @@ resource "clickhouse_clickpipe" "kafka_azure_eventhub" {`
`24`	`24`	`replicas = 1`
`25`	`25`	`}`
`26`	`26`
`27`		`- state = "Running"`
	`27`	`+`
`28`	`28`
`29`	`29`	`source = {`
`30`	`30`	`kafka = {`
Original file line number	Diff line number	Diff line change
`@@ -28,8 +28,6 @@ resource "clickhouse_clickpipe" "kafka_confluent" {`
`28`	`28`	`replicas = 1`
`29`	`29`	`}`
`30`	`30`
`31`		`- state = "Running"`
`32`		`-`
`33`	`31`	`source = {`
`34`	`32`	`kafka = {`
`35`	`33`	`type = "confluent"`
Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,7 @@ resource "clickhouse_clickpipe" "kafka_msk" {`
`23`	`23`	`replicas = 1`
`24`	`24`	`}`
`25`	`25`
`26`		`- state = "Running"`
	`26`	`+`
`27`	`27`
`28`	`28`	`source = {`
`29`	`29`	`kafka = {`
Original file line number	Diff line number	Diff line change
`@@ -28,7 +28,7 @@ resource "clickhouse_clickpipe" "kafka_msk" {`
`28`	`28`	`replicas = 1`
`29`	`29`	`}`
`30`	`30`
`31`		`- state = "Running"`
	`31`	`+`
`32`	`32`
`33`	`33`	`source = {`
`34`	`34`	`kafka = {`
Original file line number	Diff line number	Diff line change
`@@ -39,7 +39,7 @@ resource "clickhouse_clickpipe" "kafka_schema_registry" {`
`39`	`39`	`replicas = 1`
`40`	`40`	`}`
`41`	`41`
`42`		`- state = "Running"`
	`42`	`+`
`43`	`43`
`44`	`44`	`source = {`
`45`	`45`	`kafka = {`
Original file line number	Diff line number	Diff line change
`@@ -32,7 +32,7 @@ resource "clickhouse_clickpipe" "multiple" {`
`32`	`32`	`replica_memory_gb = 1.0`
`33`	`33`	`}`
`34`	`34`
`35`		`- state = "Running"`
	`35`	`+`
`36`	`36`
`37`	`37`	`source = {`
`38`	`38`	`kafka = {`