import { EventEmitter } from "events" ; import * as util from "util" ; import { IConsumer , IPartition , IPartitionWithEpoch , IQueuedMessage , IZookeeperClient , } from "@fluidframework/server-services-core" ; import { ZookeeperClient } from "@fluidframework/server-services-ordering-zookeeper" ; import * as kafka from "kafka-node" ; import { ensureTopics } from "./kafkaTopics" ; const defaultReconnectDelay = 5000 ; /**
 * Kafka consumer using the kafka-node library
 */ export class KafkaNodeConsumer implements IConsumer { private client : kafka . KafkaClient ; private consumerGroup : kafka . ConsumerGroup ; private readonly events = new EventEmitter ( ) ; private readonly zookeeper : IZookeeperClient ; constructor ( private readonly clientOptions , clientId , public readonly groupId , public readonly topic , private readonly zookeeperEndpoint ? , private readonly topicPartitions ? , private readonly topicReplicationFactor ? , private readonly reconnectDelay = defaultReconnectDelay ) { clientOptions . clientId = clientId ; this . connect ( ) ; if ( zookeeperEndpoint ) { this . zookeeper = new ZookeeperClient ( zookeeperEndpoint ) ; } } public isConnected ( ) { return this . client ? true : false ; } /**
     * Returns the offset of the latest consumsed message
     */ public getLatestMessageOffset ( partitionId ) { return undefined ; } public async commitCheckpoint ( partitionId , queuedMessage ) { const commitRequest = [ { metadata : "m" , offset : queuedMessage . offset + 1 , partition : partitionId , topic : this . topic } ] ; return new Promise < void > ( ( resolve , reject ) => { this . consumerGroup . sendOffsetCommitRequest ( commitRequest , ( err , data ) => { if ( err ) { reject ( err ) ; } else { resolve ( ) ; } } ) ; } ) ; } public async close ( ) { await util . promisify ( ( ( callback ) => this . consumerGroup . close ( false , callback ) ) as any ) ( ) ; await util . promisify ( ( ( callback ) => this . client . close ( callback ) ) as any ) ( ) ; if ( this . zookeeperEndpoint ) { this . zookeeper . close ( ) ; } } public on ( event , listener ) { this . events . on ( event , listener ) ; return this ; } public once ( event , listener ) { this . events . once ( event , listener ) ; return this ; } public async pause ( ) { this . consumerGroup . pause ( ) ; } public async resume ( ) { this . consumerGroup . resume ( ) ; } private async connect ( ) { this . client = new kafka . KafkaClient ( this . clientOptions ) ; const groupId = this . groupId ; try { await ensureTopics ( this . client , [ this . topic ] , this . topicPartitions , this . topicReplicationFactor ) ; } catch ( error ) { if ( this . client ) { this . client . close ( ) ; this . client = undefined ; } this . events . emit ( "error" , error ) ; setTimeout ( ( ) => { this . connect ( ) ; } , this . reconnectDelay ) ; return ; } this . consumerGroup = new kafka . ConsumerGroup ( { kafkaHost : this . clientOptions . kafkaHost , ssl : this . clientOptions . sslOptions , sslOptions : this . clientOptions . sslOptions , id : this . clientOptions . clientId , autoCommit : false , fetchMaxBytes : 1024 * 1024 , fetchMinBytes : 1 , fromOffset : "latest" , groupId , maxTickMessages : 100000 } , [ this . topic ] ) ; this . consumerGroup . on ( "connect" , ( ) => { this . events . emit ( "connected" ) ; } ) ; this . consumerGroup . on ( "rebalancing" , ( ) => { const payloads = ( this . consumerGroup as any ) . topicPayloads ; this . events . emit ( "rebalancing" , this . getPartitions ( payloads ) ) ; } ) ; this . consumerGroup . on ( "rebalanced" , async ( ) => { const payloads = ( this . consumerGroup as any ) . topicPayloads ; const partitions = this . getPartitions ( payloads ) ; let partitionsWithEpoch ; try { partitionsWithEpoch = await this . fetchPartitionEpochs ( partitions ) ; } catch ( err ) { this . events . emit ( "error" , err ) ; } this . events . emit ( "rebalanced" , partitionsWithEpoch ) ; } ) ; this . consumerGroup . on ( "message" , ( message ) => { this . events . emit ( "data" , message ) ; } ) ; this . consumerGroup . on ( "error" , ( error ) => { this . events . emit ( "error" , error ) ; } ) ; this . consumerGroup . on ( "offsetOutOfRange" , ( error ) => { this . events . emit ( "error" , error ) ; } ) ; } private getPartitions ( rawPartitions ) { return rawPartitions . map ( ( partition ) => ( { offset : parseInt ( partition . offset , 10 ) , partition : parseInt ( partition . partition , 10 ) , topic : partition . topic } ) ) ; } private async fetchPartitionEpochs ( partitions ) { let epochs ; if ( this . zookeeperEndpoint ) { const epochsP = new Array < Promise < number > > ( ) ; for ( const partition of partitions ) { epochsP . push ( this . zookeeper . getPartitionLeaderEpoch ( this . topic , partition . partition ) ) ; } epochs = await Promise . all ( epochsP ) ; } else { epochs = new Array ( partitions . length ) . fill ( 0 ) ; } const partitionsWithEpoch = [ ] ; for ( let i = 0 ; i < partitions . length ; ++ i ) { const partitionWithEpoch = partitions [ i ] as IPartitionWithEpoch ; partitionWithEpoch . leaderEpoch = epochs [ i ] ; partitionsWithEpoch . push ( partitionWithEpoch ) ; } return partitionsWithEpoch ; } }