<?php
// producer
require_once 'vendor/autoload.php';

use longlang\phpkafka\Producer\ProducerConfig;
use longlang\phpkafka\Producer\Producer;

$config = new ProducerConfig();
$config->setBootstrapServers('localhost:9092');

$producer = new Producer($config);
$stdin = fopen('php://stdin', 'r');
while (true) {
    echo "Enter message to send (or 'exit' to quit): ";
    $message = trim(fgets($stdin));
    if ($message === 'exit') {
        break;
    }

    $producer->send('test-topic', $message);
    echo "Message sent: $message\n";    
}
fclose($stdin);



// consumer
<?php
require_once 'vendor/autoload.php';

use longlang\phpkafka\Consumer\ConsumerConfig;
use longlang\phpkafka\Consumer\Consumer;

$config = new ConsumerConfig();
$config->setBootstrapServers('localhost:9092');
$config->setGroupId('test-group');
$config->setTopics(['test-topic']);
$consumer = new Consumer($config);

while(true) {
    // retrieve messages from the Kafka topic
    $messages = $consumer->consume();
    foreach ($messages as $message) {
        echo "Received message: " . $message->getValue() . "\n";
    }

    if (empty($messages)) {
        // no messages received, wait for a while before polling again
        sleep(1);
    }
}

<?php

/**
 * A minimal Kafka consumer sample for training purposes (PHP).
 *
 * It connects to a Kafka broker running on localhost:9092, subscribes to
 * "test-topic", and continuously polls for new records, printing each one
 * to the console.
 *
 * This uses the pure-PHP "longlang/phpkafka" library (no native extension
 * required, so it works out of the box on Windows).
 *
 * Prerequisites:
 *  - PHP 7.2+
 *  - Composer dependencies installed:
 *      composer install
 *  - Kafka broker running and reachable at localhost:9092
 *  - The topic to consume from (default: "test-topic"). Create it with:
 *      kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
 *
 * Run:
 *      php consumer.php
 */

require __DIR__ . '/vendor/autoload.php';

use longlang\phpkafka\Consumer\Consumer;
use longlang\phpkafka\Consumer\ConsumerConfig;
use longlang\phpkafka\Consumer\ConsumeMessage;

const BOOTSTRAP_SERVERS = '127.0.0.1:9092';
const TOPIC_NAME = 'test-topic';
const GROUP_ID = 'training-consumer-group';

// 1. Build the consumer configuration
$config = new ConsumerConfig();
$config->setBootstrapServer(BOOTSTRAP_SERVERS);
$config->setTopic([TOPIC_NAME]);
$config->setGroupId(GROUP_ID);
$config->setGroupInstanceId(GROUP_ID . '-instance-1');

// This library has no explicit "auto.offset.reset" option: a brand-new
// consumer group with no committed offset simply starts from offset 0 for
// each partition, which is effectively the same as Kafka's "earliest".

// Poll interval, in milliseconds.
$config->setInterval(200);

// Flush output immediately instead of letting PHP buffer it — on some
// Windows terminals buffered output can make it look like nothing new is
// arriving when messages are actually being received.
ob_implicit_flush(true);
if (ob_get_level() > 0) {
    ob_end_flush();
}

echo "Subscribed to topic '" . TOPIC_NAME . "'. Waiting for messages... (Ctrl+C to stop)" . PHP_EOL;

// 2. Create the consumer. The constructor requires a callback, but we
// drive the poll loop ourselves below (instead of calling $consumer->start())
// so we can print a liveness heartbeat and see exactly what's happening on
// every poll — useful while diagnosing "no new messages" issues. The
// callback passed here is unused since consume() (unlike start()) never
// invokes it.
$consumer = new Consumer($config, function (ConsumeMessage $message) {
    // unused - see note above
});

function printMessage(ConsumeMessage $message): void
{
    printf(
        '[%s] topic=%s, partition=%d, key=%s, value=%s' . PHP_EOL,
        date('H:i:s'),
        $message->getTopic(),
        $message->getPartition(),
        $message->getKey(),
        $message->getValue()
    );
}

// 3. Manual poll loop (equivalent to $consumer->start(), but with a
// visible heartbeat every ~5 seconds so we can tell it's still polling
// even when no new records show up). Wrapped in try/catch so any
// broker/session error is printed instead of failing silently.
$pollCount = 0;
$emptyPollCount = 0;
$lastHeartbeatPrint = microtime(true);
$autoCommit = $config->getAutoCommit();

try {
    while (true) {
        $message = $consumer->consume();
        ++$pollCount;

        if (null === $message) {
            ++$emptyPollCount;
            usleep((int) ($config->getInterval() * 1000));
        } else {
            $emptyPollCount = 0;
            printMessage($message);
            if ($autoCommit) {
                $consumer->ack($message);
            }
        }

        if (microtime(true) - $lastHeartbeatPrint >= 5) {
            printf(
                '[%s] ...still polling (poll #%d, %d empty in a row)' . PHP_EOL,
                date('H:i:s'),
                $pollCount,
                $emptyPollCount
            );
            $lastHeartbeatPrint = microtime(true);
        }
    }
} catch (\Throwable $e) {
    fwrite(STDERR, 'Consumer stopped with an error: ' . $e->getMessage() . PHP_EOL);
    fwrite(STDERR, $e->getTraceAsString() . PHP_EOL);
    exit(1);
}

package com.example;

import java.util.Properties;
import java.util.Scanner;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

public class Producer {
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String TOPIC_NAME = "test-topic";
    public static void main(String[] args) {
        // 1. Build the producer configuration
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // "all" waits for every in-sync replica to acknowledge the write before
        // considering it successful — the safest setting for a training demo.
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        // 2. Create the producer
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {

            System.out.println("Connected. Type a message and press Enter to send it to '" + TOPIC_NAME + "'.");
            System.out.println("Type 'exit' to quit.");

            Scanner scanner = new Scanner(System.in);
            int messageKey = 0;

            while (true) {
                System.out.print("> ");
                String line = scanner.nextLine();

                if ("exit".equalsIgnoreCase(line)) {
                    break;
                }

                // 3. Build and send a record. Sending is asynchronous; the callback
                // is invoked once the broker acknowledges (or fails) the write.
                ProducerRecord<String, String> record =
                        new ProducerRecord<>(TOPIC_NAME, String.valueOf(messageKey++), line);

                producer.send(record, (RecordMetadata metadata, Exception exception) -> {
                    if (exception != null) {
                        System.err.println("Failed to send message: " + exception.getMessage());
                    } else {
                        System.out.printf(
                                "Sent to topic=%s, partition=%d, offset=%d%n",
                                metadata.topic(),
                                metadata.partition(),
                                metadata.offset()
                        );
                    }
                });
            }

            // 4. Make sure any buffered messages are sent before shutting down
            producer.flush();
            System.out.println("Producer closed.");
        }
    }

}

package com.example;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class Consumer {
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String TOPIC_NAME = "test-topic";
    private static final String GROUP_ID = "training-consumer-group";

    public static void main(String[] args) {
        // 1. Build the consumer configuration
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // Where to start reading if there is no committed offset yet.
        // "earliest" = read from the beginning of the topic, "latest" = only new messages.
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // Let the client auto-commit offsets periodically (fine for a training demo;
        // production consumers often commit manually for more control).
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

        // 2. Create the consumer
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {

            // 3. Subscribe to the topic(s)
            consumer.subscribe(Collections.singletonList(TOPIC_NAME));

            System.out.println("Subscribed to topic '" + TOPIC_NAME + "'. Waiting for messages... (Ctrl+C to stop)");

            // Make sure Ctrl+C shuts things down cleanly and wakes up the poll loop
            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                System.out.println("Shutting down consumer...");
                consumer.wakeup();
            }));

            // 4. Poll loop: keep asking the broker for new records
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf(
                        "topic=%s, partition=%d, offset=%d, key=%s, value=%s%n",
                        record.topic(),
                        record.partition(),
                        record.offset(),
                        record.key(),
                        record.value()
                    );
                }
            }
        } catch (org.apache.kafka.common.errors.WakeupException e) {
            // Expected when the shutdown hook calls consumer.wakeup() — safe to ignore.
            System.out.println("Consumer wakeup called, exiting.");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

"""
A minimal Kafka producer sample for training purposes (Python).
Connects to a Kafka broker running on localhost:9092 and sends messages
typed at the console to the "test-topic" topic. Pair it with consumer.py
to see messages flow end to end.
Uses the "kafka-python" library.
Prerequisites:
    pip install -r requirements.txt
    Kafka broker running and reachable at localhost:9092. Create the topic
    with:
        kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
Run:
    python producer.py
"""

from kafka import KafkaProducer
from kafka.errors import KafkaError

BOOTSTRAP_SERVERS = "localhost:9092"
TOPIC_NAME = "test-topic"


def main() -> None:
    # 1. Create the producer.
    #    - Kafka only stores bytes, so keys/values are encoded to utf-8
    #      manually below rather than via a key_serializer/value_serializer
    #      (kafka-python only accepts those as proper Serializer subclasses
    #      now, not plain lambdas).
    #    - acks="all" waits for every in-sync replica to acknowledge the write
    #      before considering it successful — the safest setting for a demo.
    producer = KafkaProducer(
        bootstrap_servers=BOOTSTRAP_SERVERS,
        acks="all",
    )

    print(f"Connected. Type a message and press Enter to send it to '{TOPIC_NAME}'.")
    print("Type 'exit' to quit.")

    message_key = 0

    try:
        while True:
            line = input("> ")

            if line.lower() == "exit":
                break

            try:
                # 2. Send the record. send() is asynchronous and returns a
                # future; get(timeout=10) blocks until the broker acknowledges
                # (or the send fails), which is convenient for this demo.
                future = producer.send(
                    TOPIC_NAME,
                    key=str(message_key).encode("utf-8"),
                    value=line.encode("utf-8"),
                )
                metadata = future.get(timeout=10)

                print(
                    f"Sent to topic={metadata.topic}, "
                    f"partition={metadata.partition}, offset={metadata.offset}"
                )
                message_key += 1
            except KafkaError as e:
                print(f"Failed to send message: {e}")
    except (EOFError, KeyboardInterrupt):
        pass
    finally:
        # 3. Make sure any buffered messages are sent before shutting down.
        producer.flush()
        producer.close()
        print("Producer closed.")


if __name__ == "__main__":
    main()

"""
A minimal Kafka consumer sample for training purposes (Python).
Connects to a Kafka broker running on localhost:9092, subscribes to
"test-topic", and continuously polls for new records, printing each one to
the console.
Uses the "kafka-python" library.
Prerequisites:
    pip install -r requirements.txt

    Kafka broker running and reachable at localhost:9092. Create the topic
    with:
        kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
Run:
    python consumer.py
"""

from datetime import datetime
from kafka import KafkaConsumer

BOOTSTRAP_SERVERS = "localhost:9092"
TOPIC_NAME = "test-topic"
# Note: use a group ID distinct from other languages' demo consumers
# (e.g. the PHP sample). Kafka clients from different libraries can
# register incompatible group-assignment protocols; sharing a group ID
# across them can raise InconsistentGroupProtocolError.
GROUP_ID = "python-training-consumer-group"

def main() -> None:
    # 1. Create the consumer.
    #    - Kafka only stores bytes, so keys/values are decoded manually
    #      below rather than via a key_deserializer/value_deserializer
    #      (kafka-python only accepts those as proper Deserializer
    #      subclasses now, not plain lambdas).
    #    - auto_offset_reset="earliest" means a brand-new consumer group
    #      starts reading from the beginning of the topic instead of only
    #      new messages.
    #    - enable_auto_commit lets the client auto-commit offsets
    #      periodically (fine for a training demo; production consumers
    #      often commit manually for more control).
    consumer = KafkaConsumer(
        TOPIC_NAME,
        bootstrap_servers=BOOTSTRAP_SERVERS,
        group_id=GROUP_ID,
        auto_offset_reset="earliest",
        enable_auto_commit=True,
        auto_commit_interval_ms=1000,
    )

    print(f"Subscribed to topic '{TOPIC_NAME}'. Waiting for messages... (Ctrl+C to stop)")

    # 2. Poll loop: keep asking the broker for new records.
    try:
        while True:
            # poll() returns a dict of {TopicPartition: [records]}; an empty
            # dict just means nothing new arrived within the timeout.
            records = consumer.poll(timeout_ms=1000)

            for topic_partition, messages in records.items():
                for message in messages:
                    timestamp = datetime.now().strftime("%H:%M:%S")
                    key = message.key.decode("utf-8") if message.key is not None else None
                    value = message.value.decode("utf-8")
                    print(
                        f"[{timestamp}] topic={message.topic}, "
                        f"partition={message.partition}, offset={message.offset}, "
                        f"key={key}, value={value}"
                    )
    except KeyboardInterrupt:
        print("Shutting down consumer...")
    finally:
        consumer.close()


if __name__ == "__main__":
    main()

services:

  kafka-1:
    image: apache/kafka:4.3.1
    container_name: kafka-1
    hostname: kafka-1
    restart: unless-stopped

    ports:
      - "19092:19092"

    environment:
      # -----------------------------
      # KRaft
      # -----------------------------
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller

      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093

      # -----------------------------
      # Listeners
      # -----------------------------
      # INTERNAL  -> Docker network
      # CONTROLLER -> KRaft controller communication
      # EXTERNAL  -> Host machine
      KAFKA_LISTENERS: INTERNAL://:9092,CONTROLLER://:9093,EXTERNAL://:19092

      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-1:9092,EXTERNAL://localhost:19092

      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT

      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER

      # -----------------------------
      # Cluster ID
      # -----------------------------
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

      # -----------------------------
      # Replication
      # -----------------------------
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_MIN_INSYNC_REPLICAS: 2

      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3

      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2

      # -----------------------------
      # General
      # -----------------------------
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

    volumes:
      - kafka-1-data:/var/lib/kafka/data

    networks:
      - kafka-net


  kafka-2:
    image: apache/kafka:4.3.1
    container_name: kafka-2
    hostname: kafka-2
    restart: unless-stopped

    ports:
      - "29092:29092"

    environment:
      # -----------------------------
      # KRaft
      # -----------------------------
      KAFKA_NODE_ID: 2
      KAFKA_PROCESS_ROLES: broker,controller

      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093

      # -----------------------------
      # Listeners
      # -----------------------------
      KAFKA_LISTENERS: INTERNAL://:9092,CONTROLLER://:9093,EXTERNAL://:29092

      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-2:9092,EXTERNAL://localhost:29092

      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT

      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER

      # -----------------------------
      # Cluster ID
      # -----------------------------
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

      # -----------------------------
      # Replication
      # -----------------------------
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_MIN_INSYNC_REPLICAS: 2

      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3

      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2

      # -----------------------------
      # General
      # -----------------------------
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

    volumes:
      - kafka-2-data:/var/lib/kafka/data

    networks:
      - kafka-net


  kafka-3:
    image: apache/kafka:4.3.1
    container_name: kafka-3
    hostname: kafka-3
    restart: unless-stopped

    ports:
      - "39092:39092"

    environment:
      # -----------------------------
      # KRaft
      # -----------------------------
      KAFKA_NODE_ID: 3
      KAFKA_PROCESS_ROLES: broker,controller

      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093

      # -----------------------------
      # Listeners
      # -----------------------------
      KAFKA_LISTENERS: INTERNAL://:9092,CONTROLLER://:9093,EXTERNAL://:39092

      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-3:9092,EXTERNAL://localhost:39092

      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT

      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER

      # -----------------------------
      # Cluster ID
      # -----------------------------
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

      # -----------------------------
      # Replication
      # -----------------------------
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_MIN_INSYNC_REPLICAS: 2

      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3

      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2

      # -----------------------------
      # General
      # -----------------------------
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

    volumes:
      - kafka-3-data:/var/lib/kafka/data

    networks:
      - kafka-net


  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    container_name: kafka-ui
    hostname: kafka-ui
    restart: unless-stopped

    depends_on:
      - kafka-1
      - kafka-2
      - kafka-3

    ports:
      - "8080:8080"

    environment:
      KAFKA_CLUSTERS_0_NAME: local-kafka

      # Kafka UI is inside Docker,
      # therefore it must use INTERNAL listeners.
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka-1:9092,kafka-2:9092,kafka-3:9092

    networks:
      - kafka-net


volumes:
  kafka-1-data:
  kafka-2-data:
  kafka-3-data:


networks:
  kafka-net:
    driver: bridge