Wikimedia edit streams as Kafka Producer and Consumer

Note :This post is a continuation of our previous article. For additional background and context, please refer to theses post here:

1. Introduction to Apache Kafka
2. Starting Kafka CLI
3. Basics of Kafka Producer & Consumer in Java
4. Kafka Consumer in Consumer Group
5. Wikimedia edit streams as Kafka Producer and Consumer
6. Kafka Advance
7. Kafka Advance API

Ever wondered how you could tap into the live pulse of Wikipedia — tracking every edit as it happens across millions of pages? In this blog, we’ll show you how to harness the power of Wikimedia’s EventStreams and turn it into a robust, real-time data pipeline using Apache Kafka — all built in Java.

We’ll walk through:

• 📥 Streaming live edit events from Wikimedia’s public Server-Sent Events (SSE) feed
• 🚀 Publishing those events into a Kafka topic using a custom Java producer
• 🔄 Consuming and processing those events using a Kafka consumer

Wikimedia offers a modern streaming service via EventStreams, which delivers real-time updates using Server-Sent Events (SSE).

Endpoint:

https://stream.wikimedia.org/v2/stream/recentchange

1. Create a Change Handler

Create a Java class that listens to the Wikimedia EventStreams feed and handles edit events.You need a library that supports Server-Sent Events (SSE).We will use OKHTTP3.

Sample Output:

package io.conduktor.demos.kafka.wikimedia;

import com.launchdarkly.eventsource.*;

import org.apache.kafka.clients.producer.*;
import org.slf4j.LoggerFactory;
import org.slf4j.Logger;

public class WikimediaChangeHandler implements EventHandler {

    KafkaProducer<String,String> kafkaProducer;
    String topic;
    private static Logger log =  LoggerFactory.getLogger(WikimediaChangeHandler.class.getSimpleName());

    public WikimediaChangeHandler(KafkaProducer<String,String> kafkaProducer, String topic){
        this.kafkaProducer = kafkaProducer;
        this.topic = topic;
    }

    @Override
    public void onOpen() {
        log.info("Stream opened");
    }

    @Override
    public void onClosed() {
        log.info("Stream closed");
        kafkaProducer.close();
    }

    @Override
    public void onMessage(String s, MessageEvent messageEvent) throws Exception {
        log.info("Received message: " + messageEvent.getData());
        // Here you can send the message to Kafka
        kafkaProducer.send( new ProducerRecord<>(topic, messageEvent.getData()));
    }

    @Override
    public void onComment(String comment) {
        log.info("Received comment: " + comment);
    }

    @Override
    public void onError(Throwable error) {
        log.error("Error occurred: " + error.getMessage());
    }

}

2. Use the Change Handler to create a Kafka Producer

Now, let's create a Kafka Producer using the Change Event Handler

package io.conduktor.demos.kafka.wikimedia;

import com.launchdarkly.eventsource.*;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.net.URI;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class WikimediaChangesProducer {
    public static void main(String[] args) throws InterruptedException {
        String bootstrapServers = "localhost:9092";

        // Create Producer Properties
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //Create the producer
        KafkaProducer<String,String> producer = new KafkaProducer<>(properties);
        String topic =  "wikimedia.recentchange";
        EventHandler eventHandler = new WikimediaChangeHandler(producer, topic);
        String url = "https://stream.wikimedia.org/v2/stream/recentchange";
        EventSource.Builder eventSourceBuilder = new EventSource.Builder(eventHandler, URI.create(url));
        EventSource eventSource = eventSourceBuilder.build();

        //start the producer in another thread
        eventSource.start();

        // we can produce for 10 minutes and block the program until then
        TimeUnit.MINUTES.sleep(10);

    }
}

Wikimedia EventStreams (SSE) ➝ WikimediaChangeHandler ➝ Kafka Producer ➝ Kafka Topic

3. Lets Run Kafka Consumer Using CLI :

1. First, run the broker on CLI

kafka-server-start /opt/homebrew/etc/kafka/kraft/server.properties

2. Then create the Kafka topic on CLI

kafka-topics - create - topic wikimedia.recentchange - bootstrap-server localhost:9092 - partitions 1 - replication-factor 1

3. Then run the Kafka-consumer on CLI :

kafka-console-consumer - bootstrap-server localhost:9092 - topic wikimedia.recentchange - from-beginning

4. The run the Producer by running WikimediaChangesProducer.java main method

You will see Producer producing data :

[okhttp-eventsource-events-[]-0] INFO WikimediaChangeHandler - Received message: {"$schema":"/mediawiki/recentchange/1.0.0","meta":{"uri":"https://hi.wikipedia.org/wiki/2020_%E0%A4%89%E0%A4%A4%E0%A5%8D%E0%A4%A4%E0%A4%B0-%E0%A4%AA%E0%A5%82%E0%A4%B0%E0%A5%8D%E0%A4%B5%E0%A5%80_%E0%A4%A6%E0%A4%BF%E0%A4%B2%E0%A5%8D%E0%A4%B2%E0%A5%80_%E0%A4%95%E0%A5%87_%E0%A4%A6%E0%A4%82%E0%A4%97%E0%A5%87","request_id":"2e8ffe72-7e7f-4b25-85b5-

And on the Consumer side data :

0%B9%88%E0%B8%A1%E0%B8%B5_1_%E0%B8%A3%E0%B8%B2%E0%B8%A2%E0%B8%81%E0%B8%B2%E0%B8%A3","comment":"เพิ่ม [[:escaiguem]] เพิ่มเข้าหมวดหมู่","timestamp":1744412160,"user":"OctraBot","bot":true,"notify_url":"https://th.wiktionary.org/w/index.php?diff=3375369&oldid=0","server_url":"https://th.wiktionary.org","server_name":"th.wiktionary.org","server_script_path":"/w","wiki":"thwiktionary","parsedcomment":"เพิ่ม <a href=\"/wiki/escaiguem\" title=\"escaiguem\">escaiguem</a> เพิ่มเข้าหมวดหมู่"}

Improving the Kafka Producer using proper properties

The Kafka producer with the right properties is crucial for achieving the desired balance between reliability, throughput, and latency. Let’s delve into each of these properties in Java with detailed explanations and examples.

1. acks (Acknowledgements)

This property controls how many brokers must receive and acknowledge a message before the producer considers the send operation successful. It directly impacts the durability of your data.

acks=0

• **acks=0 (Fire and Forget):** The producer sends the message and doesn't wait for any acknowledgement from the broker. This offers the highest throughput and lowest latency but provides the weakest durability guarantee. If the broker goes down immediately after receiving the message (or even before fully receiving it), the message will be lost.

Properties props = new Properties(); 
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "0");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");
producer.send(record); // Producer doesn't wait for confirmation producer.close();

• **acks=1 (Leader Acknowledgement):** The producer waits for acknowledgement only from the leader of the partition. If the leader receives the message and then crashes before the followers can replicate it, the message might be lost. This provides a better durability guarantee than acks=0 with a slight trade-off in throughput and latency.

acks=1

Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("acks", "1");
 KafkaProducer<String, String> producer = new KafkaProducer<>(props);
 ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");
 producer.send(record); // Producer waits for leader's confirmation producer.close();

• **acks=all or acks=-1 (Full ISR Acknowledgement):** The producer waits for acknowledgement from all in-sync replicas (ISRs) of the partition. This provides the strongest durability guarantee as the message is considered committed only when all ISRs have acknowledged it. However, it comes with the lowest throughput and highest latency.

acks=all

Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("acks", "all");
 KafkaProducer<String, String> producer = new KafkaProducer<>(props);
 ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");
 producer.send(record); // Producer waits for all ISRs to confirm producer.close();

2. min.insync.replicas (Broker Configuration)

This is a broker-level setting that works in conjunction with acks=all on the producer. It specifies the minimum number of in-sync replicas (including the leader) that must be available for a partition to accept new messages.

min.insync.replicas

• If acks is set to all on the producer, and the number of ISRs falls below min.insync.replicas, the broker will refuse to accept new messages for that partition. This prevents data loss scenarios where a write is considered successful even if not enough replicas have it.

Example Scenario: Consider a topic with a replication factor of 3 and min.insync.replicas set to 2.

• If the leader and one follower are alive (2 ISRs), the producer with acks=all will succeed.
• If only the leader is alive (1 ISR), the producer with acks=all will encounter an exception (e.g., NotEnoughReplicasException).
• Configuration in server.properties:

min.insync.replicas=2

Impact on Producer: While the producer doesn’t directly set this property, it needs to be aware of the broker’s min.insync.replicas setting when using acks=all. Setting acks=all without a sufficient min.insync.replicas can lead to producer errors and application downtime if brokers fail.

3. Kafka Topic Availability

Ensuring the target Kafka topic exists and is healthy is fundamental for successful producer operations.

• Topic Creation: The producer can be configured to auto-create topics if auto.create.topics.enable is set to true on the brokers. However, it's generally recommended to pre-create topics with the desired number of partitions and replication factor for better control and predictability.
• Topic Health: A healthy topic has a leader for each partition and a sufficient number of in-sync replicas (as governed by the replication factor and min.insync.replicas). If a topic or its partitions are unavailable (e.g., due to broker failures), the producer will not be able to send messages to those partitions.
• Producer Behavior: When a topic or partition is unavailable, the producer will typically:
• Throw exceptions (e.g., LeaderNotAvailableException).
• Retry sending the message based on the retry configuration (explained later).
• Block indefinitely if retries are not configured or exhausted.

Best Practices:

• Pre-create topics with appropriate configurations.
• Monitor the health of your Kafka cluster and topics.
• Implement proper exception handling in your producer application to manage temporary unavailability.

4. Producer Retries (retries)

This property specifies the number of times the producer will attempt to resend a message if the initial send request fails. Failures can occur due to transient network issues, leader election, or other temporary broker problems.

Default Value: The default value is often 0 or a small number. Increasing this can improve the reliability of message delivery in the face of transient errors.

Potential Issues:

• Message Duplication: If a send operation partially succeeds (e.g., the message is written to the broker but the producer doesn’t receive an acknowledgement before a timeout and retries), it can lead to duplicate messages in the topic. This is where the Idempotent Producer (explained later) becomes crucial.
• Message Ordering: Retries without idempotency can also lead to out-of-order messages if retries happen on different partitions or if the original send eventually succeeds after a retry.

Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("retries", 3); // Retry up to 3 times  
 KafkaProducer<String, String> producer = new KafkaProducer<>(props); 
// ... 
//send record ... 
 producer.close();

5. Producer Timeouts

Several timeout-related properties influence how long the producer will wait for various operations. Configuring these appropriately is essential to prevent indefinite blocking and handle failures gracefully.

Producer Timeouts

• **request.timeout.ms:** This specifies the maximum time the producer will wait for an acknowledgement from the broker for a send request. If the timeout is exceeded, the producer will consider the request as failed and potentially retry (if retries is greater than 0). The default is typically 30000 (30 seconds).
• **delivery.timeout.ms:** This property sets the maximum time a message send operation can take, including retries. If this timeout is reached before the message is successfully sent (acknowledged according to the acks setting), the producer will throw an exception. This helps prevent unbounded blocking in case of persistent issues. The default is typically 120000 (2 minutes).
• **metadata.fetch.timeout.ms:** This property defines the timeout for fetching metadata about brokers, topics, and partitions. The producer needs this metadata to route messages correctly. The default is typically 60000 (1 minute).

Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("request.timeout.ms", 10000); // Wait up to 10 seconds for acknowledgement
 props.put("delivery.timeout.ms", 60000); // Total delivery timeout of 1 minute
 KafkaProducer<String, String> producer = new KafkaProducer<>(props); // ... send record ... producer.close();

6. Idempotent Producer (enable.idempotence=true)

This crucial property ensures that the producer sends exactly one copy of each message, even if retries are enabled. It prevents the problem of duplicate messages caused by retries after a partial failure.

How it Works: When idempotence is enabled, the producer assigns a unique Producer ID (PID) and a sequence number to each message. The broker uses this information to detect and discard duplicate messages from the same producer.

Requirements: For idempotence to work correctly, the following conditions must be met:

• enable.idempotence must be set to true on the producer.
• acks must be set to all (or -1), as idempotence relies on the broker's ability to track the state of all in-sync replicas.
• retries should typically be greater than 0 to allow for recovery from transient errors.
• max.in.flight.requests.per.connection should be less than or equal to 5 (this is a safety measure to maintain ordering with retries).

Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("enable.idempotence", true);
 props.put("acks", "all");
 props.put("retries", 3);
 props.put("max.in.flight.requests.per.connection", 5); // Optional, but recommended  
KafkaProducer<String, String> producer = new KafkaProducer<>(props); 
// ... send record ... producer.close();

dempotent Producer

Benefits: Exactly-once semantics for messages sent within a single producer session to a single partition.

Limitations: Idempotence works within a single producer session. If the producer application restarts, a new PID will be assigned, and the guarantee no longer holds across sessions. For end-to-end exactly-once semantics, you need to consider Kafka Transactions.

7. Kafka Producer Default Properties

Understanding the default values of key producer properties is important as they provide a baseline behavior. While the defaults might be suitable for some basic use cases, you often need to adjust them for production environments.

Common Default Values (Subject to Kafka version):

• acks: 1
• retries: 0
• request.timeout.ms: 30000 (30 seconds)
• delivery.timeout.ms: 120000 (2 minutes)
• metadata.fetch.timeout.ms: 60000 (1 minute)
• enable.idempotence: false
• transactional.id: null (Transactions are disabled by default)
• compression.type: none
• batch.size: 16384 (16KB)
• linger.ms: 0
• buffer.memory: 33554432 (32MB)

Importance of Overriding Defaults: For production systems, you will likely need to override these defaults to:

• Improve durability (acks=all).
• Handle transient errors (retries > 0).
• Prevent data loss (enable.idempotence=true with acks=all).
• Optimize performance (batching, compression).
• Set appropriate timeouts.

8. Kafka Producer Message Compression (compression.type)

Compressing messages before sending them to the broker can significantly improve network bandwidth utilization and reduce the storage space required on the brokers. The broker will store the messages in the compressed format, and consumers will decompress them.

Kafka Producer Message Compression

Available Compression Types:

• **none (Default):** No compression is used.
• **gzip:** Good compression ratio but higher CPU overhead.
• **snappy:** Lower compression ratio than gzip but lower CPU overhead and faster. Often a good balance.
• **lz4:** Higher compression ratio than snappy and generally faster for decompression.
• **zstd:** Offers a good balance of compression ratio and speed, often better than gzip and snappy.

Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("compression.type", "snappy"); // Enable Snappy compression  
KafkaProducer<String, String> producer = new KafkaProducer<>(props); // ... send record ... producer.close();

Considerations:

• CPU Overhead: Compression and decompression consume CPU resources on both the producer and the consumer. Choose a compression type that balances the trade-off between bandwidth savings and CPU usage.
• End-to-End: Compression is end-to-end. The producer compresses, the broker stores compressed data, and the consumer decompresses.
• Batching: Compression works best with larger batches of messages.