Azure Functions Kafka trigger in NodeJS programming model V4

Programming model V4 for NodeJS Azure Functions has been out for about two years now, and overall I think it is a great improvement over V3. If you are trying to make a Kafka trigger however, you will find that support for it has declined. It is still possible to configure a Kafka trigger, but the method to do so is a bit more involved than for other triggers. It is also poorly documented, to the point that one could be fooled into thinking that the option does not exist at all. The MS Learn page for the Kafka trigger only mentions programming model V3, a quick Google search reveals nothing, and ChatGPT also fails to come up with the correct solution.

How to create the trigger

There are examples out there, if you know where to look. The Azure Functions Kafka Extension GitHub has a lot of information and examples that don’t appear on the Microsoft Learn pages, including a lot of examples for programming model V4. It turns out that while there is no dedicated function for a Kafka trigger, we can use a generic trigger instead. Here’s an example taken from the GitHub repository. You’ll notice that the configuration fields look very similar to those in the function.json of V3:

// confluent
app.generic("kafkaTriggerMany", {
   trigger: {
     type: "kafkaTrigger",
     direction: "in",
     name: "event",
     topic: "topic",
     brokerList: "%BrokerList%",
     username: "%ConfluentCloudUserName%",
     password: "%ConfluentCloudPassword%",
     consumerGroup: "$Default",
     protocol: "saslSsl",
     authenticationMode: "plain",
     dataType: "string",
     cardinality: "MANY"
   },
   handler: kafkaTriggerMany,
 });

How to make it reusable

The app.generic() configuration is not particularly intuitive, and repeating it in every Kafka-triggered function quickly becomes tedious. Fortunately, the programmatic nature of programming model V4 enables us to create a dedicated Kafka trigger ourselves. The V4 triggers don’t care where in your code they are declared, as long as they get called at function startup. This means we can write a reusable wrapper function around app.generic(), and put it in a library. We can write a generic one that supports any possible Kafka configuration, or we can write one that is tailored to our specific needs. I will give an example of the latter.

My current use case is replacing a number of Event Hubs triggers with Kafka triggers using the Kafka-compatible endpoint. With programming model V3, this would require a bit of a code change in each function because the data format differs between the Event Hub and the Kafka Trigger. The Event Hub trigger passes JSON objects to my function, while the Kafka trigger passes wrapper objects with a ‘Value’ property that contain a JSON string value. With programming model V4, I can just abstract this away in my library:

interface KafkaEvent {
    Offset: number;
    Partition: number;
    Topic: string;
    Timestamp: number;
    Value: string;
}

export function eventHubKafkaTrigger<T,R>(
  options: {
    name: string,
    namespaceName: string,
    eventHubName: string,
    consumerGroup: string,
    connectionString: string,
    cardinality: 'ONE' | 'MANY',
    handler: (data: T, context: InvocationContext) => Promise<R>,
    output?: FunctionOutput
  }
) {
  const handler = async (data: KafkaEvent | KafkaEvent[], context: InvocationContext) => {
    let parsedData: T
      //If you copy this code, probably add some error handling here
      if (options.cardinality === 'MANY') {
        parsedData = (data as KafkaEvent[]).map(event => JSON.parse(event.Value)) as T
      } else {
        parsedData = JSON.parse((data as KafkaEvent).Value)
      }
    return options.handler(parsedData, context)
  }
  app.generic(options.name, {
    trigger: {
      type: 'kafkaTrigger',
      direction: 'in',
      name: 'event',
      topic: options.eventHubName,
      brokerList: `${options.namespaceName}.servicebus.windows.net:9093`,
      username: '$ConnectionString',
      password: options.connectionString,
      consumerGroup: options.consumerGroup,
      protocol: 'saslSsl',
      authenticationMode: 'plain',
      dataType: 'string',
      cardinality: options.cardinality
    },
    handler,
    return: options.output
  })
}

Conclusion

It would be easy for Microsoft to provide a dedicated Kafka trigger by creating their own wrapper function around app.generic(), and maybe one day they will do so. In the meantime, the examples in the Azure Functions Kafka Extension GitHub repo should enable you write your own.

If you have any questions or remarks, please leave a comment. Happy coding!