Improve env validation and definition

configure.ts

@@ -56,11 +56,20 @@ export async function configure(command: ConfigureCommand) {
   await codemods.defineEnvValidations({
     variables: {
       KAFKA_BROKERS: `KafkaEnv.schema.brokers()`,
+
       KAFKA_CLIENT_ID: `Env.schema.string.optional()`,
-      KAFKA_GROUP_ID: `Env.schema.string.optional()`,
+      KAFKA_LOG_LEVEL: `Env.schema.enum.optional(['fatal', 'error', 'warn', 'info', 'debug', 'trace'])`,
+
       KAFKA_CONNECTION_TIMEOUT: `Env.schema.number.optional()`,
       KAFKA_REQUEST_TIMEOUT: `Env.schema.number.optional()`,
-      KAFKA_LOG_LEVEL: `Env.schema.enum.optional(['fatal', 'error', 'warn', 'info', 'debug', 'trace'])`,
+      KAFKA_AUTHENTICATION_TIMEOUT: `Env.schema.number.optional()`,
+      KAFKA_REAUTHENTICATION_TIMEOUT: `Env.schema.number.optional()`,
+
+      // We don't yet support AWS IAM or OAuth Bearer or custom mechanisms
+      KAFKA_SECURITY_PROTOCOL: `Env.schema.enum.optional(['PLAIN', 'SSL'])`,
+      KAFKA_SASL_MECHANISM: `Env.schema.enum.optional(['PLAIN', 'SCRAM-SHA-256', 'SCRAM-SHA-512'])`,
+      KAFKA_SASL_USERNAME: `Env.schema.string.optional()`,
+      KAFKA_SASL_PASSWORD: `Env.schema.string.optional()`,
     },
     leadingComment: 'Variables for configuring kafka package',
   })

src/types.ts

@@ -4,7 +4,8 @@ import type {
   ConsumerRunConfig as KafkaConsumerRunConfig,
   Message as KafkaMessage,
   EachMessagePayload as KafkaEachMessagePayload,
-  SASLOptions,
+  SASLOptions as KafkaSASLOptions,
+  OauthbearerProviderResponse as KafkaOauthbearerProviderResponse,
 } from 'kafkajs'
 
 import type tls from 'node:tls'
@@ -15,6 +16,23 @@ import type { Producer } from './producer.ts'
 
 import { Kafka } from './index.ts'
 
+// AWS mechanism isn't currently well supported:
+type SASLAWSOption = { mechanism: 'aws' } & {
+  authorizationIdentity: string
+  accessKeyId: string
+  secretAccessKey: string
+  sessionToken?: string
+}
+
+// OAuth Bearer mechanism isn't currently well supported:
+type SASLOAuthOption = {
+  mechanism: 'oauthbearer'
+} & {
+  oauthBearerProvider: () => Promise<KafkaOauthbearerProviderResponse>
+}
+
+type SASLOptions = Exclude<KafkaSASLOptions, SASLAWSOption | SASLOAuthOption>
+
 export type ProducerConfig = KafkaProducerConfig
 
 export type ConsumerGroupConfig = KafkaConsumerConfig &

stubs/config/kafka.stub

@@ -3,11 +3,30 @@
 }}}
 import env from '#start/env'
 
+const kafkaSecurityProtocol = env.get('KAFKA_SECURITY_PROTOCOL', 'PLAIN')
+const useSSL = kafkaSecurityProtocol === 'SSL'
+const useSasl = env.get('KAFKA_SASL_MECHANISM') !== undefined
+
+const saslOptions = useSasl
+  ? {
+      mechanism: env.get('KAFKA_SASL_MECHANISM')?.toLowerCase(),
+      username: env.get('KAFKA_SASL_USERNAME'),
+      password: env.get('KAFKA_SASL_PASSWORD'),
+    }
+  : undefined
+
 const kafkaConfig = {
   brokers: env.get('KAFKA_BROKERS', 'localhost:9092'),
-  clientId: env.get('KAFKA_CLIENT_ID'),
+  ssl: useSSL,
+  sasl: saslOptions,
+  clientId: env.get('KAFKA_CLIENT_ID', 'local'),
+  timeouts: {
+    connection: env.get('KAFKA_CONNECTION_TIMEOUT'),
+    authentication: env.get('KAFKA_AUTHENTICATION_TIMEOUT'),
+    reauthentication: env.get('KAFKA_REAUTHENTICATION_TIMEOUT'),
+    request: env.get('KAFKA_REQUEST_TIMEOUT'),
+  },
   logLevel: env.get('KAFKA_LOG_LEVEL', env.get('LOG_LEVEL')),
 }
 
 export default kafkaConfig
-