diff --git a/README.md b/README.md
index 40e12197..fb1ae282 100644
--- a/README.md
+++ b/README.md
@@ -17,6 +17,11 @@ provider "kafka-connect" {
   url = "http://localhost:8083"
   basic_auth_username = "user" # Optional
   basic_auth_password = "password" # Optional
+  
+  # For TLS
+  tls_auth_crt = "/tmp/cert.pem" # Optional
+  tls_auth_key = "/tmp/key.pem " # Optional
+  tls_auth_is_insecure = true   # Optionnal if you do not want to check CA   
 }
 
 resource "kafka-connect_connector" "sqlite-sink" {
@@ -40,11 +45,14 @@ resource "kafka-connect_connector" "sqlite-sink" {
 
 ## Provider Properties
 
-| Property              | Type              | Example                 | Alternative environment variable name |
-|-----------------------|-------------------|-------------------------|---------------------------------------|
-| `url`                 | URL               | "http://localhost:8083" | `KAFKA_CONNECT_URL`                   |
-| `basic_auth_username` | String            | "user"                  | `KAFKA_CONNECT_BASIC_AUTH_USERNAME`   |
-| `basic_auth_password` | String            | "password"              | `KAFKA_CONNECT_BASIC_AUTH_PASSWORD`   |
+| Property              | Type   | Example                 | Alternative environment variable name |
+|-----------------------|--------|-------------------------|---------------------------------------|
+| `url`                 | URL    | "http://localhost:8083" | `KAFKA_CONNECT_URL`                   |
+| `basic_auth_username` | String | "user"                  | `KAFKA_CONNECT_BASIC_AUTH_USERNAME`   |
+| `basic_auth_password` | String | "password"              | `KAFKA_CONNECT_BASIC_AUTH_PASSWORD`   |
+| `tls_auth_crt`        | String | "certificate"           | `KAFKA_CONNECT_TLS_AUTH_CRT`          |
+| `tls_auth_key`        | String | "Key"                   | `KAFKA_CONNECT_TLS_AUTH_KEY`          |
+| `tls_auth_is_insecure`| String | "Key"                   | `KAFKA_CONNECT_TLS_IS_INSECURE`       |
 | `headers`             | Map[String]String | {foo = "bar"}           | N/A                                   |
 
 ## Resource Properties
diff --git a/connect/provider.go b/connect/provider.go
index 2e423114..88fecd29 100644
--- a/connect/provider.go
+++ b/connect/provider.go
@@ -2,6 +2,7 @@ package connect
 
 import (
 	"context"
+ 	"crypto/tls"
 	"log"
 
 	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
@@ -29,6 +30,20 @@ func Provider() *schema.Provider {
 				Optional:    true,
 				DefaultFunc: schema.EnvDefaultFunc("KAFKA_CONNECT_BASIC_AUTH_PASSWORD", ""),
 			},
+			"tls_auth_crt": {
+				Type:        schema.TypeString,
+				Optional:    true,
+				DefaultFunc: schema.EnvDefaultFunc("KAFKA_CONNECT_TLS_AUTH_CRT", ""),
+			},
+			"tls_auth_key": {
+				Type:        schema.TypeString,
+				Optional:    true,
+				DefaultFunc: schema.EnvDefaultFunc("KAFKA_CONNECT_TLS_AUTH_KEY", ""),
+			},
+			"tls_auth_is_insecure": {
+				Type:        schema.TypeBool,
+				Optional:    true,
+				DefaultFunc: schema.EnvDefaultFunc("KAFKA_CONNECT_TLS_IS_INSECURE", ""),
 			"headers": {
 				Type: schema.TypeMap,
 				Elem: &schema.Schema{
@@ -58,6 +73,23 @@ func providerConfigure(ctx context.Context, d *schema.ResourceData) (interface{}
 		c.SetBasicAuth(user, pass)
 	}
 
+	crt := d.Get("tls_auth_crt").(string)
+	key := d.Get("tls_auth_key").(string)
+	is_insecure := d.Get("tls_auth_is_insecure").(bool)
+	log.Printf("[INFO]Cert : %s\nKey: %s", crt, key)
+	log.Printf("[INFO]SSl connection is insecure : %t", is_insecure)
+
+	if crt != "" && key != "" {
+		cert, err := tls.LoadX509KeyPair(crt, key)
+		if err != nil {
+			log.Fatalf("client: loadkeys: %s", err)
+		} else {
+			if is_insecure {
+				c.SetInsecureSSL()
+			}
+			c.SetClientCertificates(cert)	
+		}
+	}
 	headers := d.Get("headers").(map[string]interface{})
 	if headers != nil {
 		for k, v := range headers {