Add support for Kafka version configuration

Fixes #329

Signed-off-by: Adrien Fillon <[email protected]>

README.md

@@ -101,6 +101,17 @@ provider "kafka" {
   sasl_aws_region   = "us-east-1"
 }
 ```
+#### Compatibility with Redpanda
+
+```hcl
+provider "kafka" {
+  bootstrap_servers = ["localhost:9092"]
+  kafka_version = "2.1.0"
+}
+```
+
+Due to Redpanda not implementing some Metadata APIs, we need to force the Kafka version to use when creating the provider.
+>>>>>>> d9b1386 (Add support for Kafka version configuration)
 
 | Property                | Description                                                                                                           | Default    |
 | -------------------     | --------------------------------------------------------------------------------------------------------------------- | ---------- |
@@ -109,6 +120,7 @@ provider "kafka" {
 | `client_cert`           | The client certificate or path to a file containing the client certificate in `PEM` format. Use for Client authentication to Kafka.<br>If you have Intermediate CA certificate(s) append them to `client_cert`.| `""`       |
 | `client_key`            | The private key or path to a file containing the private key that the client certificate was issued for.              | `""`       |
 | `client_key_passphrase` | The passphrase for the private key that the certificate was issued for.                                               | `""`       |
+| `kafka_version`         | The version of Kafka protocol to use in `$MAJOR.$MINOR.$PATCH` format. Some features may not be available on older versions.  | `""`       |
 | `tls_enabled`           | Enable communication with the Kafka Cluster over TLS.                                                                 | `true`     |
 | `skip_tls_verify`       | Skip TLS verification.                                                                                                | `false`    |
 | `sasl_username`         | Username for SASL authentication.                                                                                     | `""`       |

docs/data-sources/topic.md

@@ -25,5 +25,3 @@ description: |-
 - `id` (String) The ID of this resource.
 - `partitions` (Number) Number of partitions.
 - `replication_factor` (Number) Number of replicas.
-
-

docs/index.md

@@ -28,6 +28,7 @@ description: |-
 - `client_key` (String) The private key that the certificate was issued for.
 - `client_key_file` (String, Deprecated) Path to a file containing the private key that the certificate was issued for.
 - `client_key_passphrase` (String) The passphrase for the private key that the certificate was issued for.
+- `kafka_version` (String) The version of Kafka protocol to use in `$MAJOR.$MINOR.$PATCH` format. Some features may not be available on older versions.
 - `sasl_mechanism` (String) SASL mechanism, can be plain, scram-sha512, scram-sha256, aws-iam
 - `sasl_password` (String) Password for SASL authentication.
 - `sasl_username` (String) Username for SASL authentication.

docs/resources/acl.md

@@ -31,5 +31,3 @@ description: |-
 ### Read-Only
 
 - `id` (String) The ID of this resource.
-
-

docs/resources/quota.md

@@ -27,5 +27,3 @@ description: |-
 ### Read-Only
 
 - `id` (String) The ID of this resource.
-
-

docs/resources/topic.md

@@ -28,5 +28,3 @@ description: |-
 ### Read-Only
 
 - `id` (String) The ID of this resource.
-
-

docs/resources/user_scram_credential.md

@@ -28,5 +28,3 @@ description: |-
 ### Read-Only
 
 - `id` (String) The ID of this resource.
-
-

kafka/config.go

@@ -24,6 +24,7 @@ type Config struct {
 	ClientCert              string
 	ClientCertKey           string
 	ClientCertKeyPassphrase string
+	KafkaVersion            string
 	TLSEnabled              bool
 	SkipTLSVerify           bool
 	SASLUsername            string
@@ -95,7 +96,17 @@ func (c *Config) Token() (*sarama.AccessToken, error) {
 
 func (c *Config) newKafkaConfig() (*sarama.Config, error) {
 	kafkaConfig := sarama.NewConfig()
-	kafkaConfig.Version = sarama.V2_7_0_0
+
+	if c.KafkaVersion != "" {
+		version, err := sarama.ParseKafkaVersion(c.KafkaVersion)
+		if err != nil {
+			return kafkaConfig, fmt.Errorf("error parsing kafka version '%s': %w", c.KafkaVersion, err)
+		}
+		kafkaConfig.Version = version
+	} else {
+		kafkaConfig.Version = sarama.V2_7_0_0
+	}
+
 	kafkaConfig.ClientID = "terraform-provider-kafka"
 	kafkaConfig.Admin.Timeout = time.Duration(c.Timeout) * time.Second
 	kafkaConfig.Metadata.Full = true // the default, but just being clear
@@ -275,6 +286,7 @@ func (config *Config) copyWithMaskedSensitiveValues() Config {
 		config.ClientCert,
 		"*****",
 		"*****",
+		config.KafkaVersion,
 		config.TLSEnabled,
 		config.SkipTLSVerify,
 		config.SASLUsername,

kafka/provider.go

@@ -66,6 +66,11 @@ func Provider() *schema.Provider {
 				Optional:    true,
 				DefaultFunc: schema.EnvDefaultFunc("KAFKA_SASL_IAM_AWS_REGION", nil),
 				Description: "AWS region where MSK is deployed.",
+			"kafka_version": {
+				Type:        schema.TypeString,
+				Optional:    true,
+				DefaultFunc: schema.EnvDefaultFunc("KAFKA_VERSION", "2.7.0"),
+				Description: "The version of Kafka protocol to use in `$MAJOR.$MINOR.$PATCH` format. Some features may not be available on older versions. Default is 2.7.0.",
 			},
 			"sasl_aws_role_arn": {
 				Type:        schema.TypeString,
@@ -160,6 +165,7 @@ func providerConfigure(d *schema.ResourceData) (interface{}, error) {
 		ClientCert:              d.Get("client_cert").(string),
 		ClientCertKey:           d.Get("client_key").(string),
 		ClientCertKeyPassphrase: d.Get("client_key_passphrase").(string),
+		KafkaVersion:            d.Get("kafka_version").(string),
 		SkipTLSVerify:           d.Get("skip_tls_verify").(bool),
 		SASLAWSRegion:           d.Get("sasl_aws_region").(string),
 		SASLUsername:            d.Get("sasl_username").(string),