diff --git a/worker/server_state.go b/worker/server_state.go
index 5607903304d..1e6e5f12611 100644
--- a/worker/server_state.go
+++ b/worker/server_state.go
@@ -42,7 +42,7 @@ const (
 	BadgerDefaults = `compression=snappy; numgoroutines=8;`
 	CacheDefaults  = `size-mb=1024; percentage=0,65,35;`
 	CDCDefaults    = `file=; kafka=; sasl_user=; sasl_password=; ca_cert=; client_cert=; ` +
-		`client_key=; sasl-mechanism=PLAIN;`
+		`client_key=; sasl-mechanism=PLAIN; tls=false;`
 	GraphQLDefaults = `introspection=true; debug=false; extensions=true; poll-interval=1s; ` +
 		`lambda-url=;`
 	LimitDefaults = `mutations=allow; query-edge=1000000; normalize-node=10000; ` +
diff --git a/worker/sink_handler.go b/worker/sink_handler.go
index b43b261b7fa..bc25783825e 100644
--- a/worker/sink_handler.go
+++ b/worker/sink_handler.go
@@ -86,8 +86,18 @@ func newKafkaSink(config *z.SuperFlag) (Sink, error) {
 	saramaConf.Producer.Return.Successes = true
 	saramaConf.Producer.Return.Errors = true
 
-	if config.GetPath("ca-cert") != "" {
-		tlsCfg := &tls.Config{}
+	if config.GetBool("tls") && config.GetPath("ca-cert") == "" {
+		tlsCfg := x.TLSBaseConfig()
+		var pool *x509.CertPool
+		var err error
+		if pool, err = x509.SystemCertPool(); err != nil {
+			return nil, err
+		}
+		tlsCfg.RootCAs = pool
+		saramaConf.Net.TLS.Enable = true
+		saramaConf.Net.TLS.Config = tlsCfg
+	} else if config.GetPath("ca-cert") != "" {
+		tlsCfg := x.TLSBaseConfig()
 		var pool *x509.CertPool
 		var err error
 		if pool, err = x509.SystemCertPool(); err != nil {
diff --git a/x/tls_helper.go b/x/tls_helper.go
index 27536f8ff9a..df5e7722e1c 100644
--- a/x/tls_helper.go
+++ b/x/tls_helper.go
@@ -263,11 +263,34 @@ func setupClientAuth(authType string) (tls.ClientAuthType, error) {
 	return tls.NoClientCert, nil
 }
 
+// TLSBaseConfig returns a *tls.Config with the base set of security
+// requirements (minimum TLS v1.2 and set of cipher suites)
+func TLSBaseConfig() *tls.Config {
+	tlsCfg := new(tls.Config)
+	tlsCfg.MinVersion = tls.VersionTLS12
+	tlsCfg.CipherSuites = []uint16{
+		tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
+		tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
+		tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
+		tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
+		tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305,
+		tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305,
+		tls.TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA,
+		tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA,
+		tls.TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA,
+		tls.TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA,
+		tls.TLS_RSA_WITH_AES_128_GCM_SHA256,
+		tls.TLS_RSA_WITH_AES_256_GCM_SHA384,
+		tls.TLS_RSA_WITH_AES_256_CBC_SHA,
+	}
+	return tlsCfg
+}
+
 // GenerateServerTLSConfig creates and returns a new *tls.Config with the
 // configuration provided.
 func GenerateServerTLSConfig(config *TLSHelperConfig) (tlsCfg *tls.Config, err error) {
 	if config.CertRequired {
-		tlsCfg = new(tls.Config)
+		tlsCfg = TLSBaseConfig()
 		cert, err := tls.LoadX509KeyPair(config.Cert, config.Key)
 		if err != nil {
 			return nil, err
@@ -286,23 +309,6 @@ func GenerateServerTLSConfig(config *TLSHelperConfig) (tlsCfg *tls.Config, err e
 		}
 		tlsCfg.ClientAuth = auth
 
-		tlsCfg.MinVersion = tls.VersionTLS12
-		tlsCfg.CipherSuites = []uint16{
-			tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
-			tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
-			tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
-			tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
-			tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305,
-			tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305,
-			tls.TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA,
-			tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA,
-			tls.TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA,
-			tls.TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA,
-			tls.TLS_RSA_WITH_AES_128_GCM_SHA256,
-			tls.TLS_RSA_WITH_AES_256_GCM_SHA384,
-			tls.TLS_RSA_WITH_AES_256_CBC_SHA,
-		}
-
 		return tlsCfg, nil
 	}
 	return nil, nil