kafka_producer.bal

kafka:ProducerConfiguration messagePersistConfig = {
    clientId: "message-persist-client",
    acks: "1",
    retryCount: 3
};
final kafka:Producer messagePersistProducer = check new ("localhost:9092", messagePersistConfig);

public isolated function publishWeatherNotification(string location, WeatherReport weatherReport) returns error? {
    json payload = weatherReport.toJson();
    byte[] serializedContent = payload.toJsonString().toBytes();
    check messagePersistProducer->send({
        topic: location,
        value:  serializedContent
    });
    check messagePersistProducer->'flush();
}