Feature data source kafka topic list + Fix acceptance test

README.md

@@ -225,12 +225,65 @@ resource "kafka_user_scram_credential" "test" {
 
 #### Properties
 
-| Property             | Description                                    |
-| -------------------- | ---------------------------------------------- |
-| `username`        | The username                         |
-| `scram_mechanism`        | The SCRAM mechanism (SCRAM-SHA-256 or SCRAM-SHA-512)          |
-| `scram_iterations`             | The number of SCRAM iterations (must be >= 4096). Default: 4096       |
-| `password` | The password for the user |
+| Property            | Description                                                     |
+|---------------------|-----------------------------------------------------------------|
+| `username`          | The username                                                    |
+| `scram_mechanism`   | The SCRAM mechanism (SCRAM-SHA-256 or SCRAM-SHA-512)            |
+| `scram_iterations`  | The number of SCRAM iterations (must be >= 4096). Default: 4096 |
+| `password`          | The password for the user                                       |
+
+## Data Sources
+### `kafka_topic`
+
+A data source for getting information about a kafka topic.
+
+#### Example
+
+```hcl
+provider "kafka" {
+  bootstrap_servers = ["localhost:9092"]
+}
+data "kafka_topic" "test" {
+  name = "test_topic_name"
+}
+output "output_test" {
+  value = data.kafka_topic.test.partitions
+```
+
+### `kafka_topics`
+
+A data source for getting a list of all available Kafka topics. The object topic return will have the same property of the kafka_topic data source
+#### Example
+
+```hcl
+provider "kafka" {
+  bootstrap_servers = ["localhost:9092"]
+}
+data "kafka_topics" "test" {
+}
+output "output_test_topic_name" {
+  value = data.kafka_topics.test[0].name
+}
+output "output_test_topic_partitions" {
+  value = data.kafka_topics.test[0].partitions
+}
+output "output_test_topic_replication_factor" {
+  value = data.kafka_topics.test[0].replication_factor
+}
+output "output_test_topic_config" {
+  value = data.kafka_topics.test[0].config["retention.ms"]
+}
+```
+
+#### Properties
+
+| Property            | Description                                                     |
+|---------------------|-----------------------------------------------------------------|
+| `list`              | The list containing all kafka topics                            |
+| `username`          | The username                                                    |
+| `scram_mechanism`   | The SCRAM mechanism (SCRAM-SHA-256 or SCRAM-SHA-512)            |
+| `scram_iterations`  | The number of SCRAM iterations (must be >= 4096). Default: 4096 |
+| `password`          | The password for the user                                       |
 
 ## Requirements
 * [>= Kafka 1.0.0][3]

kafka/client.go

@@ -598,3 +598,18 @@ func (c *Client) getDeleteAclsRequestAPIVersion() int16 {
 func (c *Client) getDescribeConfigAPIVersion() int16 {
 	return int16(c.versionForKey(32, 1))
 }
+
+func (c *Client) getKafkaTopics() ([]Topic, error) {
+	topics, err := c.client.Topics()
+	if err != nil {
+		return nil, err
+	}
+	topicList := make([]Topic, len(topics))
+	for i := range topicList {
+		topicList[i], err = c.ReadTopic(topics[i], true)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return topicList, nil
+}

kafka/data_source_kafka_topics.go

@@ -0,0 +1,81 @@
+package kafka
+
+import (
+	"context"
+	"fmt"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+)
+
+func kafkaTopicsDataSource() *schema.Resource {
+	return &schema.Resource{
+		ReadContext: dataSourceTopicsRead,
+		Schema: map[string]*schema.Schema{
+			"list": &schema.Schema{
+				Type:        schema.TypeList,
+				Computed:    true,
+				Description: "A list containing all the topics.",
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"topic_name": &schema.Schema{
+							Type:        schema.TypeString,
+							Computed:    true,
+							Description: "The name of the topic.",
+						},
+						"partitions": &schema.Schema{
+							Type:        schema.TypeInt,
+							Computed:    true,
+							Description: "Number of partitions.",
+						},
+						"replication_factor": &schema.Schema{
+							Type:        schema.TypeInt,
+							Computed:    true,
+							Description: "Number of replicas.",
+						},
+						"config": &schema.Schema{
+							Type:        schema.TypeMap,
+							Computed:    true,
+							Description: "A map of string k/v attributes.",
+							Elem:        schema.TypeString,
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+func dataSourceTopicsRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	var diags diag.Diagnostics
+	client := meta.(*LazyClient)
+	topicList, err := client.GetKafkaTopics()
+	if err != nil {
+		return diag.FromErr(err)
+	}
+
+	topics := flattenTopicsData(&topicList)
+	if err := d.Set("list", topics); err != nil {
+		return diag.FromErr(err)
+	}
+	if err != nil {
+		return diag.FromErr(err)
+	}
+	d.SetId(fmt.Sprint(len(topics)))
+	return diags
+}
+
+func flattenTopicsData(topicList *[]Topic) []interface{} {
+	if topicList != nil {
+		topics := make([]interface{}, len(*topicList))
+		for i, topic := range *topicList {
+			topicObj := make(map[string]interface{})
+			topicObj["topic_name"] = topic.Name
+			topicObj["replication_factor"] = topic.ReplicationFactor
+			topicObj["partitions"] = topic.Partitions
+			topicObj["config"] = topic.Config
+			topics[i] = topicObj
+		}
+		return topics
+	}
+	return make([]interface{}, 0)
+}

kafka/data_source_kafka_topics_test.go

@@ -0,0 +1,77 @@
+package kafka
+
+import (
+	"fmt"
+	"github.com/hashicorp/go-uuid"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/terraform"
+	"testing"
+
+	r "github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+)
+
+func TestAcc_Topics(t *testing.T) {
+	u, err := uuid.GenerateUUID()
+	if err != nil {
+		t.Fatal(err)
+	}
+	topicName := fmt.Sprintf("syslog-%s", u)
+
+	bs := testBootstrapServers[0]
+	r.Test(t, r.TestCase{
+		ProviderFactories: overrideProviderFactory(),
+		Steps: []r.TestStep{
+			{
+				Config: cfg(t, bs, fmt.Sprintf(testDataSourceKafkaTopics, topicName)),
+				Check: r.ComposeTestCheckFunc(
+					testDatasourceTopics,
+				),
+			},
+		},
+	})
+}
+
+const testDataSourceKafkaTopics = `
+resource "kafka_topic" "test" {
+  name               = "%[1]s"
+  replication_factor = 1
+  partitions         = 1
+  config = {
+    "retention.ms" = "22222"
+  }
+}
+data "kafka_topics" "test" {
+ depends_on = [kafka_topic.test]
+}
+`
+
+func testDatasourceTopics(s *terraform.State) error {
+	resourceState := s.Modules[0].Resources["data.kafka_topics.test"]
+	if resourceState == nil {
+		return fmt.Errorf("resource not found in state")
+	}
+	instanceState := resourceState.Primary
+	client := testProvider.Meta().(*LazyClient)
+	expectedTopics, err := client.GetKafkaTopics()
+	if err != nil {
+		return fmt.Errorf(err.Error())
+	}
+	for i := 0; i < len(expectedTopics); i++ {
+		expectedTopicName := instanceState.Attributes[fmt.Sprintf("list.%d.topic_name", i)]
+		expectedTopicOutput, err := client.ReadTopic(expectedTopicName, true)
+		if err != nil {
+			return fmt.Errorf(err.Error())
+		}
+
+		if instanceState.Attributes[fmt.Sprintf("list.%d.partitions", i)] != fmt.Sprint(expectedTopicOutput.Partitions) {
+			return fmt.Errorf("expected %d for topic %s partition, got %s", expectedTopicOutput.Partitions, expectedTopicOutput.Name, instanceState.Attributes[fmt.Sprintf("list.%d.partitions", i)])
+		}
+		if instanceState.Attributes[fmt.Sprintf("list.%d.replication_factor", i)] != fmt.Sprint(expectedTopicOutput.ReplicationFactor) {
+			return fmt.Errorf("expected %d for topic %s replication factor, got %s", expectedTopicOutput.ReplicationFactor, expectedTopicOutput.Name, instanceState.Attributes[fmt.Sprintf("list.%d.replication_factor", i)])
+		}
+		retentionMs := expectedTopicOutput.Config["retention.ms"]
+		if instanceState.Attributes[fmt.Sprintf("list.%d.config.retention.ms", i)] != *retentionMs {
+			return fmt.Errorf("expected %s for topic %s config retention.ms, got %s", *retentionMs, expectedTopicOutput.Name, instanceState.Attributes[fmt.Sprintf("list.%d.config.retention.ms", i)])
+		}
+	}
+	return nil
+}

kafka/lazy_client.go

@@ -185,3 +185,10 @@ func (c *LazyClient) DeleteUserScramCredential(userScramCredential UserScramCred
 	}
 	return c.inner.DeleteUserScramCredential(userScramCredential)
 }
+func (c *LazyClient) GetKafkaTopics() ([]Topic, error) {
+	err := c.init()
+	if err != nil {
+		return nil, err
+	}
+	return c.inner.getKafkaTopics()
+}

kafka/provider.go

@@ -107,7 +107,8 @@ func Provider() *schema.Provider {
 			"kafka_user_scram_credential": kafkaUserScramCredentialResource(),
 		},
 		DataSourcesMap: map[string]*schema.Resource{
-			"kafka_topic": kafkaTopicDataSource(),
+			"kafka_topic":  kafkaTopicDataSource(),
+			"kafka_topics": kafkaTopicsDataSource(),
 		},
 	}
 }