Allow plans when connection to Kafka cannot be established (#251)

Fixes #250 
Fixes #178 
Fixes #189

kafka/lazy_init_test.go

@@ -0,0 +1,61 @@
+package kafka
+
+import (
+	"fmt"
+	"testing"
+
+	uuid "github.com/hashicorp/go-uuid"
+	r "github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+)
+
+//lintignore:AT001
+func TestAcc_LazyInit(t *testing.T) {
+	u, err := uuid.GenerateUUID()
+	if err != nil {
+		t.Fatal(err)
+	}
+	topicName := fmt.Sprintf("syslog-%s", u)
+	bs := "localhost:90"
+
+	_, err = overrideProvider()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	r.Test(t, r.TestCase{
+		ProviderFactories: overrideProviderFactory(),
+		Steps: []r.TestStep{
+			{
+				Config:             cfg(t, bs, fmt.Sprintf(test_allResources_config, topicName, topicName)),
+				PlanOnly:           true,
+				ExpectNonEmptyPlan: true,
+			},
+		},
+	})
+}
+
+const test_allResources_config = `
+resource "kafka_topic" "test" {
+  name               = "%s"
+  replication_factor = 1
+  partitions         = 1
+}
+
+resource "kafka_acl" "test" {
+	resource_name                = "%s"
+	resource_type                = "Topic"
+	resource_pattern_type_filter = "Prefixed"
+	acl_principal                = "User:Alice"
+	acl_host                     = "*"
+	acl_operation                = "Write"
+	acl_permission_type          = "Deny"
+}
+
+resource "kafka_quota" "test" {
+  entity_name               = "my-app"
+  entity_type               = "client-id"
+  config = {
+		"producer_byte_rate" = "2500000"
+  }
+}
+`

kafka/provider_test.go

@@ -31,7 +31,7 @@ func testAccPreCheck(t *testing.T) {
 		t.Fatal("No client")
 	}
 	if err := client.init(); err != nil {
-		t.Fatalf("Bad init %v", err)
+		t.Fatalf("Client could not be initialized %v", err)
 	}
 }
 

kafka/resource_kafka_quota_test.go

@@ -184,7 +184,6 @@ provider "kafka" {
 
 const testResourceQuota1 = `
 resource "kafka_quota" "test1" {
-  provider = "kafka"
   entity_name               = "%s"
   entity_type               = "client-id"
   config = {

kafka/resource_kafka_topic.go

@@ -284,9 +284,10 @@ func topicRead(ctx context.Context, d *schema.ResourceData, meta interface{}) di
 }
 
 func customDiff(ctx context.Context, diff *schema.ResourceDiff, v interface{}) error {
-	log.Printf("[INFO] Checking the diff!")
-	client := v.(*LazyClient)
-
+	// Skip custom logic for resource creation.
+	if diff.Id() == "" {
+		return nil
+	}
 	if diff.HasChange("partitions") {
 		log.Printf("[INFO] Partitions have changed!")
 		o, n := diff.GetChange("partitions")
@@ -302,13 +303,16 @@ func customDiff(ctx context.Context, diff *schema.ResourceDiff, v interface{}) e
 	}
 
 	if diff.HasChange("replication_factor") {
+		log.Printf("[INFO] Checking the diff!")
+		client := v.(*LazyClient)
+
 		canAlterRF, err := client.CanAlterReplicationFactor()
 		if err != nil {
 			return err
 		}
 
 		if !canAlterRF {
-			log.Println("[INFO] Need kafka >= 2.4.0 to update replication_factor in-place")
+			log.Println("[INFO] Need Kafka >= 2.4.0 to update replication_factor in-place")
 			if err := diff.ForceNew("replication_factor"); err != nil {
 				return err
 			}