CentOS 7 PHP Apache Kafka: Installation, Examples, and System Diagram

This guide will walk you through installing all the necessary components on CentOS 7 to run PHP applications with Apache, and interact with Apache Kafka.

1. CentOS 7 Base System Preparation

First, ensure your system is up-to-date and install essential utilities.

# Update all existing packages
sudo yum update -y

# Install EPEL (Extra Packages for Enterprise Linux) repository
sudo yum install epel-release -y

# Install basic development tools and utilities
sudo yum install wget git nano vim unzip -y

# Install yum-utils for repository management
sudo yum install yum-utils -y

2. Java Development Kit (JDK) Installation

Apache Kafka and Zookeeper are written in Java, so a JDK is required. We'll install OpenJDK 8.

sudo yum install java-1.8.0-openjdk-devel -y

# Verify Java installation
java -version

3. Apache HTTP Server (httpd) Installation

Apache will serve your PHP applications.

# Install Apache
sudo yum install httpd -y

# Start and enable Apache to run on boot
sudo systemctl start httpd
sudo systemctl enable httpd

# Configure firewall to allow HTTP and HTTPS traffic
sudo firewall-cmd --permanent --add-service=http
sudo firewall-cmd --permanent --add-service=https
sudo firewall-cmd --reload

# Verify Apache is running (optional, check http://your_server_ip in browser)

4. PHP Installation (with remi repository)

CentOS 7's default PHP version is outdated. We'll use the remi repository for a more modern PHP version (e.g., PHP 7.4 or 8.x). PHP-devel is crucial for compiling extensions.

# Install Remi repository
sudo yum install http://rpms.remirepo.net/enterprise/remi-release-7.rpm -y

# Enable PHP 7.4 (you can replace 'remi-php74' with 'remi-php80' or 'remi-php81' for newer versions)
sudo yum-config-manager --enable remi-php74

# Install PHP and necessary modules for web and development (php-devel is for pecl)
sudo yum install php php-cli php-common php-gd php-ldap php-mbstring php-mysqlnd php-pdo php-opcache php-xml php-devel -y

# Restart Apache to load PHP module
sudo systemctl restart httpd

# Verify PHP installation
php -v

5. Apache Kafka and Zookeeper Installation

Kafka requires Zookeeper to manage its cluster state. We'll download and set up a single-node Kafka environment.

# Create a directory for Kafka
sudo mkdir /opt/kafka
cd /opt/kafka

# Download Kafka (find the latest stable version from https://kafka.apache.org/downloads)
# Example: Kafka 3.6.1 with Scala 2.13
KAFKA_VERSION="3.6.1"
SCALA_VERSION="2.13"
wget "https://downloads.apache.org/kafka/${KAFKA_VERSION}/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz"

# Extract the Kafka archive
sudo tar -xvzf kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz --strip 1

# Remove the archive
rm kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz

# Configure Zookeeper and Kafka to store logs and data in a specific location
sudo mkdir -p /var/lib/zookeeper
sudo mkdir -p /var/lib/kafka-logs

# Edit Zookeeper configuration (if needed, default is usually fine)
# For this basic setup, default zookeeper.properties is fine, but you can change dataDir if you like.
# Example: sudo nano config/zookeeper.properties
# dataDir=/var/lib/zookeeper

# Edit Kafka configuration
# sudo nano config/server.properties
# Make sure log.dirs is set
# log.dirs=/var/lib/kafka-logs
# listener and advertised.listeners might need to be set if you have multiple network interfaces
# listeners=PLAINTEXT://:9092
# advertised.listeners=PLAINTEXT://your_server_ip:9092

Create Systemd Services for Zookeeper and Kafka
This allows you to manage Zookeeper and Kafka easily with systemctl.

a. Zookeeper Service (/etc/systemd/system/zookeeper.service)

sudo nano /etc/systemd/system/zookeeper.service

Paste the following content:

[Unit]
Description=Apache Zookeeper
Documentation=http://zookeeper.apache.org
Requires=network.target remote-fs.target
After=network.target remote-fs.target

[Service]
Type=simple
ExecStart=/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties
ExecStop=/opt/kafka/bin/zookeeper-server-stop.sh
Restart=on-failure

[Install]
WantedBy=multi-user.target

b. Kafka Service (/etc/systemd/system/kafka.service)

sudo nano /etc/systemd/system/kafka.service

Paste the following content:

[Unit]
Description=Apache Kafka
Documentation=http://kafka.apache.org
Requires=zookeeper.service
After=zookeeper.service

[Service]
Type=simple
ExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
Restart=on-failure

[Install]
WantedBy=multi-user.target

Reload Systemd and Start Services

sudo systemctl daemon-reload

# Start and enable Zookeeper
sudo systemctl start zookeeper
sudo systemctl enable zookeeper
sudo systemctl status zookeeper

# Start and enable Kafka
sudo systemctl start kafka
sudo systemctl enable kafka
sudo systemctl status kafka

# Configure firewall for Kafka (default port 9092)
sudo firewall-cmd --permanent --add-port=9092/tcp
sudo firewall-cmd --reload

6. PHP rdkafka Extension Installation

The rdkafka PHP extension provides a native client for Apache Kafka, built on librdkafka.

# Install librdkafka-devel (C library development files)
sudo yum install librdkafka-devel -y

# Install rdkafka PHP extension via PECL
sudo pecl install rdkafka

# Enable the rdkafka extension in php.ini
# PECL usually asks if you want to add it to php.ini, but if not, do it manually:
echo "extension=rdkafka.so" | sudo tee /etc/php.d/40-rdkafka.ini

# Restart Apache and PHP-FPM (if using FPM, otherwise just Apache)
sudo systemctl restart httpd
# If you configured PHP-FPM, you would also restart it: sudo systemctl restart php-fpm

# Verify the extension is loaded
php -m | grep rdkafka
# You should see 'rdkafka' in the output.

Simple PHP Kafka Examples

Now that all components are installed, let's create some simple PHP scripts to interact with Kafka.

Pre-requisite: Create a Kafka topic for testing.

/opt/kafka/bin/kafka-topics.sh --create --topic my_test_topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
/opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092

1. Kafka Producer Example

This script demonstrates how to send a message to a Kafka topic.

producer.php

<?php

// Configure Kafka producer
$conf = new RdKafka\Conf();
$conf->set('bootstrap.servers', 'localhost:9092'); // Kafka broker address
$conf->set('client.id', 'php-producer'); // Identifier for this client

// If you need to handle delivery reports (optional)
$conf->setDrMsgCb(function ($kafka, $message) {
    if ($message->err) {
        echo "Message delivery failed: " . $message->errstr() . "\n";
    } else {
        echo "Message delivered to topic {$message->topic_name} [" . $message->partition . "] at offset {$message->offset}\n";
    }
});

// Create a new Kafka producer instance
$producer = new RdKafka\Producer($conf);

// Specify the topic to produce messages to
$topic = $producer->newTopic('my_test_topic');

// Message details
$key = 'my_key';
$payload = 'Hello Kafka from PHP!';

// Produce the message (RD_KAFKA_PARTITION_UA means "unassigned partition", Kafka will choose one)
$topic->produce(RD_KAFKA_PARTITION_UA, 0, $payload, $key);

// Poll for any delivery reports (important for async producers)
// The loop ensures delivery reports are processed before script exits
for ($i = 0; $i < 10; $i++) {
    $producer->poll(0); // Poll for 0 ms, non-blocking
}

// Flush any outstanding messages and wait for them to be delivered
// 1000ms (1 second) timeout
$result = $producer->flush(1000);

if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
    echo "Successfully produced message.\n";
} else {
    echo "Failed to flush outstanding messages or timed out.\n";
}

?>

Description:
This script creates an RdKafka\Producer instance, configured to connect to localhost:9092. It defines a delivery report callback to confirm if messages were successfully sent. It then produces a single message "Hello Kafka from PHP!" with a key "my_key" to the my_test_topic. The poll() and flush() calls are crucial for ensuring the message is actually sent and delivery reports are processed before the script exits.

How to run:

php producer.php

2. Kafka Consumer Example

This script demonstrates how to consume messages from a specific Kafka topic.

consumer.php

<?php

// Configure Kafka consumer
$conf = new RdKafka\Conf();
$conf->set('bootstrap.servers', 'localhost:9092');
$conf->set('group.id', 'php-consumer-group'); // Assign to a consumer group
$conf->set('auto.offset.reset', 'earliest'); // Start consuming from the beginning if no offset is stored

// Create a new Kafka consumer instance
$consumer = new RdKafka\Consumer($conf);

// Create a topic subscription and subscribe
$consumer->subscribe(['my_test_topic']);

echo "Waiting for messages... Press Ctrl+C to exit.\n";

while (true) {
    // Poll for messages with a 120-second timeout
    // The timeout is important as it allows the consumer to rebalance
    $message = $consumer->consume(120 * 1000); 
    
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            echo "Message received: \n";
            echo "  Topic: " . $message->topic_name . "\n";
            echo "  Partition: " . $message->partition . "\n";
            echo "  Offset: " . $message->offset . "\n";
            echo "  Key: " . $message->key . "\n";
            echo "  Payload: " . $message->payload . "\n";
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            echo "No more messages, waiting for new ones...\n";
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            echo "Timed out waiting for new messages...\n";
            break;
        default:
            echo "Consumer error: " . $message->errstr() . "\n";
            break;
    }
}

?>

Description:
This script creates an RdKafka\Consumer instance. It configures a group.id for the consumer, which is essential for Kafka's consumer group management. It subscribes to my_test_topic and then enters a continuous loop, polling for messages. When a message is received, its details (topic, partition, offset, key, payload) are printed. It also handles end-of-partition and timeout scenarios.

How to run:

php consumer.php

Run this script, then run producer.php in another terminal to see messages being consumed.

3. Consumer with Manual Offset Committing Example

This script shows how to manually commit offsets, giving you more control over when messages are considered “processed.”

manual_commit_consumer.php

<?php

$conf = new RdKafka\Conf();
$conf->set('bootstrap.servers', 'localhost:9092');
$conf->set('group.id', 'php-manual-commit-group');
$conf->set('auto.offset.reset', 'earliest');

// Disable auto-commit
$conf->set('enable.auto.commit', 'false');

// Optional: Set a callback for offset commit results
$conf->setOffsetCommitCb(function ($kafka, $err, $partitions) {
    if ($err === RD_KAFKA_RESP_ERR_NO_ERROR) {
        echo "Offsets committed successfully!\n";
    } else {
        echo "Error committing offsets: " . rd_kafka_err2str($err) . "\n";
        foreach ($partitions as $partition) {
            echo "  Topic: {$partition->getTopic()}, Partition: {$partition->getPartition()}, Offset: {$partition->getOffset()}\n";
        }
    }
});

$consumer = new RdKafka\Consumer($conf);
$consumer->subscribe(['my_test_topic']);

echo "Waiting for messages with manual commit... Press Ctrl+C to exit.\n";

$messagesProcessed = 0;
while (true) {
    $message = $consumer->consume(10 * 1000); // 10-second timeout
    
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            echo "Processing message: " . $message->payload . " (Offset: " . $message->offset . ")\n";
            // Simulate some work
            sleep(1); 
            $messagesProcessed++;

            // Commit offsets after every 5 messages or if a significant time passes
            if ($messagesProcessed % 5 === 0) {
                echo "Committing offsets after " . $messagesProcessed . " messages.\n";
                $consumer->commitAsync($message); // Commit up to this message's offset
                $messagesProcessed = 0; // Reset counter for next batch
            }
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            echo "Reached end of partition, waiting...\n";
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            echo "Timed out. No new messages in 10 seconds. Attempting to commit any remaining offsets.\n";
            // In a real application, you might commit here even if not 5 messages
            $consumer->commitAsync(); // Commit current position of all subscribed topics
            break;
        default:
            echo "Consumer error: " . $message->errstr() . "\n";
            break;
    }
}

?>

Description:
This example builds upon the basic consumer by explicitly disabling auto.commit and implementing manual offset committing. The commitAsync($message) method is used to commit the offset of the last processed message (plus one) to Kafka. This ensures that messages are only marked as processed after your application has successfully handled them, preventing data loss in case of a crash. The script commits offsets every 5 messages or when a timeout occurs, simulating batch processing.

How to run:

php manual_commit_consumer.php

Kafka Architecture Diagram

This diagram illustrates the basic components and how PHP interacts with Kafka.

Explanation of the Diagram:

• Client Application (CentOS 7 Server): This represents your CentOS 7 machine where PHP and Apache are installed.
  • Web Browser: Initiates requests, typically to the Apache HTTP Server.
  • Apache HTTP Server: Serves web content and forwards PHP requests to the PHP interpreter.
  • PHP-FPM / PHP CLI: The PHP interpreter executes your PHP scripts.
  • librdkafka PHP Extension (rdkafka.so): This is the PHP extension you installed, which provides the RdKafka classes and functions for your PHP code.
  • librdkafka C Library: The PHP extension is a wrapper around the high-performance librdkafka C library, which handles the actual network communication with Kafka brokers.
• Kafka Cluster:
  • Kafka Broker: The core component of Kafka. It receives messages from producers, stores them, and delivers them to consumers. Our setup has one broker.
  • Zookeeper: A distributed coordination service that Kafka uses to manage its cluster state (e.g., topic configurations, partition leaders, consumer group offsets).

Flow:
1. A PHP script (either run via CLI or invoked by Apache) uses the rdkafka PHP extension.
2. The rdkafka extension calls functions in the librdkafka C library.
3. librdkafka establishes TCP/IP connections to the Kafka Broker (default port 9092) to send messages (producer) or fetch messages (consumer).
4. The Kafka Broker communicates with Zookeeper (default port 2181) to store metadata and manage its state.