Kafka (Message queue service)

package sh.platform.languages.sample;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import sh.platform.config.Config;
import sh.platform.config.Kafka;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Supplier;

public class KafkaSample implements Supplier<String> {

    @Override
    public String get() {
        StringBuilder logger = new StringBuilder();

        // Create a new config object to ease reading the Platform.sh environment variables.
        // You can alternatively use getenv() yourself.
        Config config = new Config();

        try {
            // Get the credentials to connect to the Kafka service.
            final Kafka kafka = config.getCredential("kafka", Kafka::new);
            Map<String, Object> configProducer = new HashMap<>();
            configProducer.putIfAbsent(ProducerConfig.CLIENT_ID_CONFIG, "animals");
            final Producer<Long, String> producer = kafka.getProducer(configProducer);

            logger.append("<ul>");
            // Sending data into the stream.
            RecordMetadata metadata = producer.send(new ProducerRecord<>("animals", "lion")).get();
            logger.append("<li>Record sent with to partition <code>").append(metadata.partition())
                    .append("</code> with offset <code>").append(metadata.offset()).append("</code></li>");

            metadata = producer.send(new ProducerRecord<>("animals", "dog")).get();
            logger.append("<li>Record sent with to partition <code>").append(metadata.partition())
                    .append("</code> with offset <code>").append(metadata.offset()).append("</code></li>");

            metadata = producer.send(new ProducerRecord<>("animals", "cat")).get();
            logger.append("<li>Record sent with to partition <code>").append(metadata.partition())
                    .append("</code> with offset <code>").append(metadata.offset()).append("</code></li>");
            logger.append("</ul>");

            // Consumer, read data from the stream.
            final HashMap<String, Object> configConsumer = new HashMap<>();
            configConsumer.put(ConsumerConfig.GROUP_ID_CONFIG, "consumerGroup1");
            configConsumer.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

            Consumer<Long, String> consumer = kafka.getConsumer(configConsumer, "animals");
            ConsumerRecords<Long, String> consumerRecords = consumer.poll(Duration.ofSeconds(3));

            logger.append("<ul>");
            // Print each record.
            consumerRecords.forEach(record -> {
                logger.append("<li>Record: Key <code>" + record.key());
                logger.append("</code> value <code>" + record.value());
                logger.append("</code> partition <code>" + record.partition());
                logger.append("</code> offset <code>" + record.offset()).append("</code></li>");
            });
            logger.append("</ul>");

            // Commits the offset of record to broker.
            consumer.commitSync();

            return logger.toString();
        } catch (Exception exp) {
            throw new RuntimeException("An error when execute Kafka", exp);
        }
    }
}

<?php

/*
with rdkafka PHP extension installed, you can use the RdKafka\Producer class to produce messages to a Kafka topic:
applications:
  app:
    type: php:8.5
    runtime:
      extensions:
        - rdkafka
*/

declare(strict_types=1);

try {
    $host = getenv('KAFKA_HOST') ?: getenv('KAFKA_IP');
    $port = getenv('KAFKA_PORT');

    if (!$host || !$port) {
        throw new RuntimeException('Kafka environment variables not available.');
    }

    $broker = sprintf('%s:%s', $host, $port);

    $conf = new RdKafka\Conf();
    $conf->set('bootstrap.servers', $broker); // <-- important

    $producer = new RdKafka\Producer($conf);

    $topic = $producer->newTopic('test');

    $topic->produce(
        RD_KAFKA_PARTITION_UA, // let Kafka choose the partition
        0,
        'Hello, World!'
    );

    // Serve delivery reports / internal queue
    $producer->poll(0);

    // Wait for delivery
    $timeoutMs = 10000;
    $start = microtime(true);

    while ($producer->getOutQLen() > 0) {
        $producer->poll(100);

        if ((microtime(true) - $start) * 1000 > $timeoutMs) {
            break;
        }
    }

    $result = $producer->flush($timeoutMs);

    if ($result !== RD_KAFKA_RESP_ERR_NO_ERROR) {
        throw new RuntimeException('Unable to flush messages (not delivered within timeout).');
    }

    echo "Message delivered.\n";
} catch (\Throwable $e) {
    print $e->getMessage();
}

from json import dumps
from json import loads
from kafka import KafkaConsumer, KafkaProducer
from platformshconfig import Config


def usage_example():
    # Create a new Config object to ease reading the Platform.sh environment variables.
    # You can alternatively use os.environ yourself.
    config = Config()
    # Get the credentials to connect to the Kafka service.
    credentials = config.credentials('kafka')
    
    try:
        kafka_server = '{}:{}'.format(credentials['host'], credentials['port'])
        
        # Producer
        producer = KafkaProducer(
            bootstrap_servers=[kafka_server],
            value_serializer=lambda x: dumps(x).encode('utf-8')
        )
        for e in range(10):
            data = {'number' : e}
            producer.send('numtest', value=data)
        
        # Consumer
        consumer = KafkaConsumer(
            bootstrap_servers=[kafka_server],
            auto_offset_reset='earliest'
        )
        
        consumer.subscribe(['numtest'])
        
        output = ''
        # For demonstration purposes so it doesn't block.
        for e in range(10):
            message = next(consumer)
            output += str(loads(message.value.decode('UTF-8'))["number"]) + ', '

        # What a real implementation would do instead.
        # for message in consumer:
        #     output += loads(message.value.decode('UTF-8'))["number"]

        return output
    
    except Exception as e:
        return e

## With the ruby-kafka gem

# Producer
require "kafka"
kafka = Kafka.new(["kafka.internal:9092"], client_id: "my-application")
kafka.deliver_message("Hello, World!", topic: "greetings")

# Consumer
kafka.each_message(topic: "greetings") do |message|
  puts message.offset, message.key, message.value
end