Motivation

Chaos

Order

Concepts

Overview

Topics & Partitions

Guarantees

• Messages sent to a partition will be appended in the order they are seen
• Consumer see messages in the order they are stored in the log
• Kafka will tolerate up to n-1 server failures for a topic with replication factor n

Log Compaction

Example

Java Producer

public class News {
    public final UUID id;
    public final String author, title, body;
...
}
					

Properties config = new Properties();
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker);
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, NewsSerializer.class.getName());

KafkaProducer<String, News> producer = new KafkaProducer<>(config);
					

public RecordMetadata send(News news) throws ExecutionException, InterruptedException {
    ProducerRecord<String, News> record = new ProducerRecord<>(topic, news.id.toString(), news);
    Future<RecordMetadata> recordMetadataFuture = this.producer.send(record);
    return recordMetadataFuture.get();
}
					

Message Distribution

• Round Robin

  								ProducerRecord(String topic, V value);
  							

• Via key hash (murmur2)

  								ProducerRecord(String topic, K key, V value);
  							

• Via specific partition

  								ProducerRecord(String topic, Integer partition, K key, V value);
  							

Consumer Grouping

• Topic A:Nachrichten entweder von A0 oder A1
• Topic B:Nachrichten sowohl von B0 als auch von einem Consumer aus Gruppe C empfangen

Consumer Threading

• #Threads > #Paritions → überschüssigen Threads erhalten keine Nachrichten (siehe T0)
• #Threads < #Paritions → Threads bekommen Nachrichten von mehreren Partitionen → keine Order mehr!!

Java Consumer Thread

public void run() {
    Thread.currentThread().setName(name);
    ConsumerIterator<String, News> it = messageStream.iterator();
    while (it.hasNext()) {
        relayMessage(it.next());
    }
}
					

void relayMessage(MessageAndMetadata<String, News> kafkaMessage) {
    logger.trace("Received message with key '{}' and offset '{}' "
        + "on partition '{}' for topic '{}'",
        kafkaMessage.key(), kafkaMessage.offset(),
        kafkaMessage.partition(), kafkaMessage.topic());
    messageConsumer.consume(kafkaMessage.message());
}
					

public interface NewsConsumer<News> {
    void consume(News message);
}
					

• High Level Consumer API lagert das Verwalten des Offset nach ZooKeeper oder Kafka aus
• SimpleConsumer überlässt dies der Consumer-Implementierung
• Client Re-Design ist in Arbeit

Wiring

import static kafka.consumer.Consumer.createJavaConsumerConnector;

Properties props = new Properties();
props.put("zookeeper.connect", zookeeper); // list of ZooKeeper nodes
props.put("group.id", groupId);            // identifies consumer group
props.put("offsets.storage", "kafka");     // storage for offsets
props.put("dual.commit.enabled", "false"); // migration switch
...
ConsumerConnector consumerConnector = createJavaConsumerConnector(new ConsumerConfig(props));
					

Map<String, List<KafkaStream<String, News>>> consumerMap;
consumerMap = consumerConnector.createMessageStreams(
    ImmutableMap.of(topic, numberOfThreads),     // number of streams per topic
    new StringDecoder(null), new NewsDecoder()); // message key and value decoders

List<KafkaStream<String, News>> streams = consumerMap.get(topic);
					

// create fixed size thread pool to launch all the threads
ExecutorService pool = Executors.newFixedThreadPool(numberOfThreads);
// create consumer threads to handle the messages
for (final KafkaStream stream : streams) {
    String name = String.format("%s[%s]", topic, threadNumber++);
    pool.submit(new ConsumerThread(stream, name, consumer));
}
					

Performance

What makes Kafka fast?

What makes Kafka fast?

Fast Writes:

• linear writes
• all writes go to OS pagecache

Fast Reads:

• linear reads
• direct data transport from pagecache to socket* via sendfile()

Combined → Fast!

* Consumer without lag get messages from pagecache! More Details: http://kafka.apache.org/documentation.html#persistence

sendfile()?

Summary

Key Features & Facts

• Commit Log  Topics, Partitioning, Log Compaction, Compression & Offset Management
• Real-time      High-Throughput & Low-Latency
• Persistence   Scalable, centralized & replicated storage

Related

• Clients  Scala, Java, Python, Go, Perl, Erlang etc.
• Integration  Storm, Samza, Hadoop, ElasticSearch, Logstash, Hive etc.
• Confluent Platform  Schema Registry, REST Proxy & Camus