{"id":52,"date":"2023-12-02T11:09:00","date_gmt":"2023-12-02T16:09:00","guid":{"rendered":"https:\/\/coding101.xyz\/?p=52"},"modified":"2023-12-02T19:57:53","modified_gmt":"2023-12-03T00:57:53","slug":"kafka-primer","status":"publish","type":"post","link":"https:\/\/coding101.xyz\/?p=52","title":{"rendered":"Kafka Primer"},"content":{"rendered":"\n

This is a short article about how to setup a Kafka application up and running. <\/p>\n\n\n\n\n\n\n\n

Kafka is a messaging service, very fast. It revolves around topics of messages. Messages have a key and a payload. Typically the server listens to port 9092 TCP<\/p>\n\n\n\n

Maven<\/h2>\n\n\n\n

The following packages are needed. The client one covers producer and consumer, the streams covers streams. Look it up for the latest version. <\/p>\n\n\n\n

<<\/span>dependencies><\/span>\n<<\/span>dependency><\/span>\n<<\/span>groupId><\/span>org.<\/span>apache<\/span>.<\/span>kafka<\/span><\/<\/span>groupId><\/span>\n<<\/span>artifactId><\/span>kafka-<\/span>clients<\/<\/span>artifactId><\/span>\n<<\/span>version><\/span>3.6<\/span>.<\/span>0<\/span><\/<\/span>version><\/span>\n<\/<\/span>dependency><\/span>\n<<\/span>dependency><\/span>\n<<\/span>groupId><\/span>org.<\/span>apache<\/span>.<\/span>kafka<\/span><\/<\/span>groupId><\/span>\n<<\/span>artifactId><\/span>kafka-<\/span>streams<\/<\/span>artifactId><\/span>\n<<\/span>version><\/span>3.6<\/span>.<\/span>0<\/span><\/<\/span>version><\/span>\n<\/<\/span>dependency><\/span>\n<\/<\/span>dependencies><\/span>\n<\/pre><\/div>\n\n\n\n
Producer<\/h2>\n\n\n\n
Producer puts messages in the topic. The producer can be accessed from multiple threads. Here is the code. The key can be changed, same the payload type, with, of course, the right serializer and deserializer.<\/p>\n\n\n\n
package<\/span> org.<\/span>example<\/span>;<\/span>\n\n\nimport<\/span> org.apache.kafka.clients.consumer.ConsumerConfig<\/span>;<\/span>\nimport<\/span> org.apache.kafka.clients.producer.KafkaProducer<\/span>;<\/span>\nimport<\/span> org.apache.kafka.clients.producer.ProducerConfig<\/span>;<\/span>\nimport<\/span> org.apache.kafka.clients.producer.ProducerRecord<\/span>;<\/span>\nimport<\/span> org.apache.kafka.common.serialization.LongSerializer<\/span>;<\/span>\nimport<\/span> org.apache.kafka.common.serialization.StringSerializer<\/span>;<\/span>\nimport<\/span> org.slf4j.Logger<\/span>;<\/span>\nimport<\/span> org.slf4j.LoggerFactory<\/span>;<\/span>\n\nimport<\/span> java.util.Date<\/span>;<\/span>\nimport<\/span> java.util.Properties<\/span>;<\/span>\n\npublic<\/span> class<\/span> ProdStarter<\/span> {<\/span>\n    private<\/span> static<\/span> final<\/span> Logger log =<\/span> LoggerFactory.<\/span>getLogger<\/span>(<\/span>ProdStarter.<\/span>class<\/span>);<\/span>\n\n    public<\/span> static<\/span> void<\/span> main<\/span>(<\/span>String[]<\/span> args)<\/span> {<\/span>\n        try<\/span> {<\/span>\n            new<\/span> ProdStarter<\/span>().<\/span>start<\/span>();<\/span>\n        }<\/span> catch<\/span> (<\/span>Exception e)<\/span> {<\/span>\n            e.<\/span>printStackTrace<\/span>();<\/span>\n        }<\/span>\n    }<\/span>\n\n    private<\/span> void<\/span> start<\/span>()<\/span> throws<\/span> Exception {<\/span>\n        KafkaProducer<<\/span>Long,<\/span> String><\/span> producer =<\/span> getStringStringKafkaProducer();<\/span>\n\n        for<\/span> (<\/span>int<\/span> i =<\/span> 0<\/span>;<\/span> i <<\/span> 10<\/span>;<\/span> i++)<\/span> {<\/span>\n            long<\/span> currentTimestamp =<\/span> new<\/span> Date().<\/span>getTime<\/span>();<\/span>\n\n\n            ProducerRecord<<\/span>Long,<\/span> String><\/span> producerRecord =<\/span>\n                    new<\/span> ProducerRecord<>(<\/span>"tester1"<\/span>,<\/span> (<\/span>long<\/span>)<\/span> i,<\/span> "this is the amount for the value - "<\/span> +<\/span> i +<\/span> " - here"<\/span>);<\/span>\n\n            \/\/ send data - asynchronous<\/span>\n            producer.<\/span>send<\/span>(<\/span>producerRecord);<\/span>\n        }<\/span>\n\n\n        \/\/ flush data - synchronous<\/span>\n        producer.<\/span>flush<\/span>();<\/span>\n\n        \/\/ flush and close producer<\/span>\n        producer.<\/span>close<\/span>();<\/span>\n    }<\/span>\n\n    private<\/span> static<\/span> KafkaProducer<<\/span>Long,<\/span> String><\/span> getStringStringKafkaProducer()<\/span> {<\/span>\n        String bootstrapServers =<\/span> "127.0.0.1:9092"<\/span>;<\/span>\n\n        \/\/ setup the producer<\/span>\n        Properties properties =<\/span> new<\/span> Properties();<\/span>\n        properties.<\/span>setProperty<\/span>(<\/span>ProducerConfig.<\/span>BOOTSTRAP_SERVERS_CONFIG<\/span>,<\/span> bootstrapServers);<\/span>\n        properties.<\/span>setProperty<\/span>(<\/span>ProducerConfig.<\/span>KEY_SERIALIZER_CLASS_CONFIG<\/span>,<\/span> LongSerializer.<\/span>class<\/span>.<\/span>getName<\/span>());<\/span>\n        properties.<\/span>setProperty<\/span>(<\/span>ProducerConfig.<\/span>VALUE_SERIALIZER_CLASS_CONFIG<\/span>,<\/span> StringSerializer.<\/span>class<\/span>.<\/span>getName<\/span>());<\/span>\n        properties.<\/span>setProperty<\/span>(<\/span>ConsumerConfig.<\/span>GROUP_ID_CONFIG<\/span>,<\/span> "groupidhere"<\/span>);<\/span>\n\n        \/\/ one instance of producer is thread safe and enough for all application<\/span>\n        \/\/ better performance than opening one for each thread or message<\/span>\n        KafkaProducer<<\/span>Long,<\/span> String><\/span> producer =<\/span> new<\/span> KafkaProducer<>(<\/span>properties);<\/span>\n        return<\/span> producer;<\/span>\n    }<\/span>\n}<\/span>\n<\/pre><\/div>\n\n\n\n
Consumer<\/h2>\n\n\n\n
The example below of course is a proof of concept. <\/p>\n\n\n\n
package<\/span> org.<\/span>example<\/span>;<\/span>\n\nimport<\/span> org.apache.kafka.clients.consumer.ConsumerConfig<\/span>;<\/span>\nimport<\/span> org.apache.kafka.clients.consumer.ConsumerRecord<\/span>;<\/span>\nimport<\/span> org.apache.kafka.clients.consumer.ConsumerRecords<\/span>;<\/span>\nimport<\/span> org.apache.kafka.clients.consumer.KafkaConsumer<\/span>;<\/span>\nimport<\/span> org.apache.kafka.common.PartitionInfo<\/span>;<\/span>\nimport<\/span> org.apache.kafka.common.serialization.StringDeserializer<\/span>;<\/span>\nimport<\/span> org.slf4j.Logger<\/span>;<\/span>\nimport<\/span> org.slf4j.LoggerFactory<\/span>;<\/span>\n\nimport<\/span> java.time.Duration<\/span>;<\/span>\nimport<\/span> java.util.List<\/span>;<\/span>\nimport<\/span> java.util.Map<\/span>;<\/span>\nimport<\/span> java.util.Properties<\/span>;<\/span>\n\npublic<\/span> class<\/span> ConsumerStarter<\/span> {<\/span>\n    private<\/span> static<\/span> final<\/span> Logger log =<\/span> LoggerFactory.<\/span>getLogger<\/span>(<\/span>ProdStarter.<\/span>class<\/span>);<\/span>\n\n    public<\/span> static<\/span> void<\/span> main<\/span>(<\/span>String[]<\/span> args)<\/span> {<\/span>\n        try<\/span> {<\/span>\n            new<\/span> ConsumerStarter<\/span>().<\/span>start<\/span>();<\/span>\n        }<\/span> catch<\/span> (<\/span>Exception e)<\/span> {<\/span>\n            e.<\/span>printStackTrace<\/span>();<\/span>\n        }<\/span>\n    }<\/span>\n\n    private<\/span> void<\/span> start<\/span>()<\/span> throws<\/span> Exception {<\/span>\n        String bootstrapServers =<\/span> "127.0.0.1:9092"<\/span>;<\/span>\n\n        Properties properties =<\/span> new<\/span> Properties();<\/span>\n        properties.<\/span>setProperty<\/span>(<\/span>ConsumerConfig.<\/span>BOOTSTRAP_SERVERS_CONFIG<\/span>,<\/span> bootstrapServers);<\/span>\n        properties.<\/span>setProperty<\/span>(<\/span>ConsumerConfig.<\/span>KEY_DESERIALIZER_CLASS_CONFIG<\/span>,<\/span> StringDeserializer.<\/span>class<\/span>.<\/span>getName<\/span>());<\/span>\n        properties.<\/span>setProperty<\/span>(<\/span>ConsumerConfig.<\/span>VALUE_DESERIALIZER_CLASS_CONFIG<\/span>,<\/span> StringDeserializer.<\/span>class<\/span>.<\/span>getName<\/span>());<\/span>\n\n        \/\/ very important if for a new group id you want to start from very beginning, as<\/span>\n        \/\/ the default value is "latest" that means that even for a new group id config you still<\/span>\n        \/\/ get messages starting "now", not "earliest"<\/span>\n        properties.<\/span>setProperty<\/span>(<\/span>ConsumerConfig.<\/span>AUTO_OFFSET_RESET_CONFIG<\/span>,<\/span> "earliest"<\/span>);<\/span>\n        properties.<\/span>setProperty<\/span>(<\/span>ConsumerConfig.<\/span>GROUP_ID_CONFIG<\/span>,<\/span> "a"<\/span>);<\/span>\n\n        KafkaConsumer<<\/span>String,<\/span> String><\/span> consumer =<\/span> new<\/span> KafkaConsumer<>(<\/span>properties);<\/span>\n\n        \/\/ just list the topics to see what's there<\/span>\n        Map<<\/span>String,<\/span> List<<\/span>PartitionInfo>><\/span> topics =<\/span> consumer.<\/span>listTopics<\/span>();<\/span>\n        topics.<\/span>forEach<\/span>((<\/span>key,<\/span> partitions)<\/span> -><\/span> {<\/span>\n            System.<\/span>out<\/span>.<\/span>println<\/span>(<\/span>"key: "<\/span> +<\/span> key);<\/span>\n            partitions.<\/span>forEach<\/span>((<\/span>part)<\/span> -><\/span> {<\/span>\n                String top =<\/span> part.<\/span>topic<\/span>();<\/span>\n                System.<\/span>out<\/span>.<\/span>println<\/span>(<\/span>"topic: "<\/span> +<\/span> top);<\/span>\n            });<\/span>\n        });<\/span>\n\n        \/\/ start reading messages<\/span>\n        consumer.<\/span>subscribe<\/span>(<\/span>List.<\/span>of<\/span>(<\/span>"tester1"<\/span>));<\/span>\n        \/\/ poll for new data<\/span>\n        while<\/span>(<\/span>true<\/span>){<\/span>\n            ConsumerRecords<<\/span>String,<\/span> String><\/span> records =<\/span>\n                    consumer.<\/span>poll<\/span>(<\/span>Duration.<\/span>ofMillis<\/span>(<\/span>100<\/span>));<\/span>\n\n            for<\/span> (<\/span>ConsumerRecord<<\/span>String,<\/span> String><\/span> record :<\/span> records){<\/span>\n                System.<\/span>out<\/span>.<\/span>println<\/span>(<\/span>"Key: "<\/span> +<\/span> record.<\/span>key<\/span>()<\/span> +<\/span> ", Value: "<\/span> +<\/span> record.<\/span>value<\/span>());<\/span>\n                System.<\/span>out<\/span>.<\/span>println<\/span>(<\/span>"Partition: "<\/span> +<\/span> record.<\/span>partition<\/span>()<\/span> +<\/span> ", Offset:"<\/span> +<\/span> record.<\/span>offset<\/span>());<\/span>\n            }<\/span>\n        }<\/span>\n    }<\/span>\n}<\/span>\n<\/pre><\/div>\n\n\n\n
Streams<\/h2>\n\n\n\n
Streams are a kind of consumers of topic messages with a lot of functionality associated with them:<\/p>\n\n\n\n
    \n
  • You can handle message structure and content including changing the key and the payload<\/li>\n\n\n\n
  • Filtering<\/li>\n\n\n\n
  • Grouping<\/li>\n\n\n\n
  • At the end you can post to another topic the updated messages<\/li>\n\n\n\n
  • Use temporary storage provided by Kafka for groups, counts and other operations<\/li>\n<\/ul>\n\n\n\n
    Here is a sample code of how to register a stream against a topic and then consume the contents<\/p>\n\n\n\n
    \n
     1\n 2\n 3\n 4\n 5\n 6\n 7\n 8\n 9\n10\n11\n12\n13\n14\n15\n16\n17\n18\n19\n20\n21\n22\n23\n24\n25\n26\n27\n28\n29\n30\n31\n32\n33\n34\n35\n36\n37\n38\n39\n40\n41\n42\n43\n44\n45\n46\n47\n48\n49<\/pre><\/td>
    package<\/span> org.<\/span>example<\/span>;<\/span>\n\nimport<\/span> org.apache.kafka.common.serialization.Serdes<\/span>;<\/span>\nimport<\/span> org.apache.kafka.common.utils.Bytes<\/span>;<\/span>\nimport<\/span> org.apache.kafka.streams.KafkaStreams<\/span>;<\/span>\nimport<\/span> org.apache.kafka.streams.KeyValue<\/span>;<\/span>\nimport<\/span> org.apache.kafka.streams.StreamsBuilder<\/span>;<\/span>\nimport<\/span> org.apache.kafka.streams.StreamsConfig<\/span>;<\/span>\nimport<\/span> org.apache.kafka.streams.kstream.*<\/span>;<\/span>\nimport<\/span> org.apache.kafka.streams.state.KeyValueStore<\/span>;<\/span>\n\nimport<\/span> java.util.Arrays<\/span>;<\/span>\nimport<\/span> java.util.List<\/span>;<\/span>\nimport<\/span> java.util.Properties<\/span>;<\/span>\n\npublic<\/span> class<\/span> StreamerTest<\/span> {<\/span>\n    public<\/span> static<\/span> void<\/span> main<\/span>(<\/span>String[]<\/span> args)<\/span> {<\/span>\n        try<\/span> {<\/span>\n            new<\/span> StreamerTest<\/span>().<\/span>start<\/span>();<\/span>\n        }<\/span> catch<\/span> (<\/span>Exception e)<\/span> {<\/span>\n            e.<\/span>printStackTrace<\/span>();<\/span>\n        }<\/span>\n    }<\/span>\n\n    private<\/span> void<\/span> start<\/span>()<\/span> throws<\/span> Exception {<\/span>\n        Properties props =<\/span> new<\/span> Properties();<\/span>\n        props.<\/span>put<\/span>(<\/span>StreamsConfig.<\/span>APPLICATION_ID_CONFIG<\/span>,<\/span> "wordcount-application"<\/span>);<\/span>\n        props.<\/span>put<\/span>(<\/span>StreamsConfig.<\/span>BOOTSTRAP_SERVERS_CONFIG<\/span>,<\/span> "localhost:9092"<\/span>);<\/span>\n        props.<\/span>put<\/span>(<\/span>StreamsConfig.<\/span>DEFAULT_KEY_SERDE_CLASS_CONFIG<\/span>,<\/span> Serdes.<\/span>Long<\/span>().<\/span>getClass<\/span>());<\/span>\n        props.<\/span>put<\/span>(<\/span>StreamsConfig.<\/span>DEFAULT_VALUE_SERDE_CLASS_CONFIG<\/span>,<\/span> Serdes.<\/span>String<\/span>().<\/span>getClass<\/span>());<\/span>\n\n        StreamsBuilder builder =<\/span> new<\/span> StreamsBuilder();<\/span>\n        KStream<<\/span>Long,<\/span> String><\/span> textLines =<\/span> builder.<\/span>stream<\/span>(<\/span>"tester1"<\/span>);<\/span>\n        textLines\n                .<\/span>mapValues<\/span>(<\/span>value -><\/span> value +<\/span> " suffix"<\/span>)<\/span>\n                .<\/span>flatMapValues<\/span>(<\/span>value -><\/span> List.<\/span>of<\/span>(<\/span>value,<\/span> value +<\/span> "(a)"<\/span>))<\/span>\n                .<\/span>filter<\/span>((<\/span>key,<\/span> value)<\/span> -><\/span> key %<\/span> 2<\/span> ==<\/span> 0<\/span>)<\/span>\n                .<\/span>map<\/span>((<\/span>key,<\/span> val)<\/span> -><\/span> new<\/span> KeyValue<>(<\/span>key +<\/span> 10<\/span>,<\/span> "prefix - "<\/span> +<\/span> val))<\/span>\n                .<\/span>foreach<\/span>((<\/span>key,<\/span> value)<\/span> -><\/span> {<\/span>\n                    String message =<\/span> String.<\/span>format<\/span>(<\/span>"only value: %d >>%s<<"<\/span>,<\/span> key,<\/span> value);<\/span>\n\n                    System.<\/span>out<\/span>.<\/span>println<\/span>(<\/span>message);<\/span>\n                });<\/span>\n\n        KafkaStreams streams =<\/span> new<\/span> KafkaStreams(<\/span>builder.<\/span>build<\/span>(),<\/span> props);<\/span>\n        streams.<\/span>start<\/span>();<\/span>\n    }<\/span>\n\n}<\/span>\n<\/pre><\/td><\/tr><\/table><\/div>\n<\/div>\n\n\n\n
    Various Commands<\/h2>\n\n\n\n
    Here are a number of commands. It is easy to figure out what they do, however you need those samples to follow the model. <\/p>\n\n\n\n
    kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092\nkafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092\nkafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092\nkafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092\nkafka-topics.sh --bootstrap-server localhost:9092 --create --topic tester1  --partitions 20 --replication-factor 3\nkafka-consumer-groups.bat --list --bootstrap-server localhost:9092\nkafka-topics.bat --bootstrap-server=localhost:9092 --list\nkafka-consumer-groups.bat --bootstrap-server=localhost:9092 --list\nkafka-console-consumer.bat --bootstrap-server localhost:9092 --topic tester1\n<\/pre><\/div>\n","protected":false},"excerpt":{"rendered":"
    This is a short article about how to setup a Kafka application up and running.<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-52","post","type-post","status-publish","format-standard","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/coding101.xyz\/index.php?rest_route=\/wp\/v2\/posts\/52","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/coding101.xyz\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/coding101.xyz\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/coding101.xyz\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/coding101.xyz\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=52"}],"version-history":[{"count":15,"href":"https:\/\/coding101.xyz\/index.php?rest_route=\/wp\/v2\/posts\/52\/revisions"}],"predecessor-version":[{"id":94,"href":"https:\/\/coding101.xyz\/index.php?rest_route=\/wp\/v2\/posts\/52\/revisions\/94"}],"wp:attachment":[{"href":"https:\/\/coding101.xyz\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=52"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/coding101.xyz\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=52"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/coding101.xyz\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=52"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}