feat(kafka): add sasl auth support

Signed-off-by: Andrew Thauer <athauer@wealthsimple.com>

plugins/kafka-backend/src/config/ClusterReader.ts

@@ -15,8 +15,7 @@
  */
 
 import { Config } from '@backstage/config';
-import { ConnectionOptions } from 'tls';
-import { ClusterDetails } from '../types/types';
+import { ClusterDetails, SslConfig, SaslConfig } from '../types/types';
 
 export function getClusterDetails(config: Config[]): ClusterDetails[] {
   return config.map(clusterConfig => {
@@ -24,14 +23,13 @@ export function getClusterDetails(config: Config[]): ClusterDetails[] {
       name: clusterConfig.getString('name'),
       brokers: clusterConfig.getStringArray('brokers'),
     };
-    const sslConfig = clusterConfig.getOptional('kafka.ssl');
-    if (sslConfig) {
-      return {
-        ...clusterDetails,
-        ssl: sslConfig as ConnectionOptions,
-      };
-    }
+    const ssl = clusterConfig.getOptional('ssl') as SslConfig;
+    const sasl = clusterConfig.getOptional('sasl') as SaslConfig;
 
-    return clusterDetails;
+    return {
+      ...clusterDetails,
+      ...(ssl ? { ssl } : {}),
+      ...(sasl ? { sasl } : {}),
+    };
   });
 }

plugins/kafka-backend/src/service/KafkaApi.ts

@@ -16,7 +16,7 @@
 
 import { Kafka, SeekEntry } from 'kafkajs';
 import { Logger } from 'winston';
-import { ConnectionOptions } from 'tls';
+import { SaslConfig, SslConfig } from '../types/types';
 
 export type PartitionOffset = {
   id: number;
@@ -31,7 +31,8 @@ export type TopicOffset = {
 export type Options = {
   clientId: string;
   brokers: string[];
-  ssl?: ConnectionOptions;
+  ssl?: SslConfig;
+  sasl?: SaslConfig;
   logger: Logger;
 };
 

plugins/kafka-backend/src/types/types.ts

@@ -19,5 +19,14 @@ import { ConnectionOptions } from 'tls';
 export interface ClusterDetails {
   name: string;
   brokers: string[];
-  ssl?: ConnectionOptions;
+  ssl?: SslConfig;
+  sasl?: SaslConfig;
 }
+
+export type SslConfig = ConnectionOptions | boolean;
+
+export type SaslConfig = {
+  mechanism: 'plain' | 'scram-sha-256' | 'scram-sha-512';
+  username: string;
+  password: string;
+};