Added support for SSL connection to the Kafka cluster

plugins/kafka-backend/README.md

@@ -14,6 +14,10 @@ The name of the client to use when connecting to the cluster.
 
 A list of the brokers' host names and ports to connect to.
 
+### ssl (optional)
+
+Configure TLS connection to the Kafka cluster. The options are passed directly to [tls.connect] and used to create the TLS secure context. Normally these would include `key` and `cert`.
+
 Example:
 
 ```yaml

plugins/kafka-backend/config.d.ts

@@ -17,5 +17,10 @@ export interface Config {
   kafka?: {
     clientId: string;
     brokers: string[];
+    ssl?: {
+      ca: string[];
+      key: string;
+      cert: string;
+    };
   };
 }

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

@@ -16,6 +16,7 @@
 
 import { Kafka, SeekEntry } from 'kafkajs';
 import { Logger } from 'winston';
+import { ConnectionOptions } from 'tls';
 
 export type PartitionOffset = {
   id: number;
@@ -27,6 +28,13 @@ export type TopicOffset = {
   partitions: PartitionOffset[];
 };
 
+export type Options = {
+  clientId: string;
+  brokers: string[];
+  ssl?: ConnectionOptions;
+  logger: Logger;
+};
+
 export interface KafkaApi {
   fetchTopicOffsets(topic: string): Promise<Array<PartitionOffset>>;
   fetchGroupOffsets(groupId: string): Promise<Array<TopicOffset>>;
@@ -36,13 +44,13 @@ export class KafkaJsApiImpl implements KafkaApi {
   private readonly kafka: Kafka;
   private readonly logger: Logger;
 
-  constructor(clientId: string, brokers: string[], logger: Logger) {
-    logger.debug(
-      `creating kafka client with clientId=${clientId} and brokers=${brokers}`,
+  constructor(options: Options) {
+    options.logger.debug(
+      `creating kafka client with clientId=${options.clientId} and brokers=${options.brokers}`,
     );
 
-    this.kafka = new Kafka({ clientId, brokers });
-    this.logger = logger;
+    this.kafka = new Kafka(options);
+    this.logger = options.logger;
   }
 
   async fetchTopicOffsets(topic: string): Promise<Array<PartitionOffset>> {

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

@@ -20,6 +20,7 @@ import { Logger } from 'winston';
 import { Config } from '@backstage/config';
 import { KafkaApi, KafkaJsApiImpl } from './KafkaApi';
 import _ from 'lodash';
+import { ConnectionOptions } from 'tls';
 
 export interface RouterOptions {
   logger: Logger;
@@ -71,7 +72,10 @@ export async function createRouter(
   const clientId = options.config.getString('kafka.clientId');
   const brokers = options.config.getStringArray('kafka.brokers');
 
-  const kafkaApi = new KafkaJsApiImpl(clientId, brokers, logger);
+  const sslConfig = options.config.getOptional('kafka.ssl');
+  const ssl = sslConfig ? (sslConfig as ConnectionOptions) : undefined;
+
+  const kafkaApi = new KafkaJsApiImpl({ clientId, brokers, logger, ssl });
 
   return makeRouter(logger, kafkaApi);
 }