support TLS without authentication

README.md

@@ -33,6 +33,7 @@ A source is getting associated with a consumer group ID the first time the `Read
 | `clientID`           | A Kafka client ID.                                                                                                                                                                                           | false    | `conduit-connector-kafka` |
 | `readFromBeginning`  | Determines from whence the consumer group should begin consuming when it finds a partition without a committed offset. If this option is set to true it will start with the first message in that partition. | false    | `false`                   |
 | `groupID`            | Defines the consumer group ID.                                                                                                                                                                               | false    |                           |
+| `tls`                | Defines whether TLS is enabled.                                                                                                                                                                              | false    | `false`                   |
 | `clientCert`         | A certificate for the Kafka client, in PEM format. If provided, the private key needs to be provided too.                                                                                                    | false    |                           |
 | `clientKey`          | A private key for the Kafka client, in PEM format. If provided, the certificate needs to be provided too.                                                                                                    | false    |                           |
 | `caCert`             | The Kafka broker's certificate, in PEM format.                                                                                                                                                               | false    |                           |

common/config.go

@@ -19,6 +19,7 @@ import (
 	"errors"
 	"time"
 
+	sdk "github.com/conduitio/conduit-connector-sdk"
 	"github.com/twmb/franz-go/pkg/kgo"
 )
 
@@ -62,7 +63,8 @@ func (c Config) Validate() error {
 // TryDial tries to establish a connection to brokers and returns nil if it
 // succeeds to connect to at least one broker.
 func (c Config) TryDial(ctx context.Context) error {
-	cl, err := kgo.NewClient(kgo.SeedBrokers(c.Servers...))
+	opts := c.FranzClientOpts(sdk.Logger(ctx))
+	cl, err := kgo.NewClient(opts...)
 	if err != nil {
 		return err
 	}

common/config_test.go

@@ -57,6 +57,7 @@ func TestConfig_Validate(t *testing.T) {
 		name: "invalid Client cert",
 		cfg: Config{
 			ConfigTLS: ConfigTLS{
+				TLSEnabled: true,
 				ClientCert: "foo",
 			},
 		},
@@ -65,7 +66,8 @@ func TestConfig_Validate(t *testing.T) {
 		name: "invalid Client key",
 		cfg: Config{
 			ConfigTLS: ConfigTLS{
-				ClientKey: "foo",
+				TLSEnabled: true,
+				ClientKey:  "foo",
 			},
 		},
 		wantErr: "tls: failed to find any PEM data in certificate input",

common/tls.go

@@ -21,6 +21,8 @@ import (
 )
 
 type ConfigTLS struct {
+	// TLSEnabled defines whether TLS is needed to communicate with the Kafka cluster.
+	TLSEnabled bool `json:"tls"`
 	// ClientCert is the Kafka client's certificate.
 	ClientCert string `json:"clientCert"`
 	// ClientKey is the Kafka client's private key.
@@ -46,7 +48,7 @@ func (c ConfigTLS) TLS() *tls.Config {
 }
 
 func (c ConfigTLS) tls() (*tls.Config, error) {
-	if c.ClientCert == "" && c.CACert == "" && c.ClientKey == "" {
+	if !c.TLSEnabled {
 		return nil, nil
 	}