{"id":11208,"date":"2024-10-22T10:19:15","date_gmt":"2024-10-22T15:19:15","guid":{"rendered":"https:\/\/www.rushworth.us\/lisa\/?p=11208"},"modified":"2024-10-22T10:19:16","modified_gmt":"2024-10-22T15:19:16","slug":"kafka-consumer-and-producer-example-java","status":"publish","type":"post","link":"https:\/\/www.rushworth.us\/lisa\/?p=11208","title":{"rendered":"Kafka Consumer and Producer Example \u2013 Java"},"content":{"rendered":"\n

Producer<\/h2>\n\n\n
\npackage com.example;\n\nimport org.apache.kafka.clients.producer.KafkaProducer;\nimport org.apache.kafka.clients.producer.ProducerRecord;\nimport org.apache.kafka.clients.producer.ProducerConfig;\nimport org.apache.kafka.common.serialization.StringSerializer;\n\nimport java.io.InputStream;\nimport java.time.LocalDateTime;\nimport java.time.format.DateTimeFormatter;\nimport java.util.Properties;\n\npublic class SimpleKafkaProducer {\n\n    public static void main(String[] args) {\n        if (args.length < 2) {\n            System.err.println("Please provide Kafka server and port. Usage: SimpleKafkaProducer <kafka-server> <port>");\n            System.exit(1);\n        }\n\n        String kafkaHost = args[0];\n        String kafkaPort = args[1];\n        String kafkaServer = kafkaHost + ":" + kafkaPort;\n        String topicName = "LJRJavaTest";\n        String truststorePassword = "truststore-password";\n\n        Properties props = new Properties();\n        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);\n        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());\n        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());\n\n        \/\/ Load truststore from resources\n        try (InputStream truststoreStream = SimpleKafkaProducer.class.getResourceAsStream("\/kafka.client.truststore.jks")) {\n            if (truststoreStream == null) {\n                throw new RuntimeException("Truststore not found in resources");\n            }\n\n            \/\/ Create a temporary file to hold the truststore\n            java.nio.file.Path tempTruststore = java.nio.file.Files.createTempFile("kafka.client.truststore", ".jks");\n            java.nio.file.Files.copy(truststoreStream, tempTruststore, java.nio.file.StandardCopyOption.REPLACE_EXISTING);\n\n            props.put("security.protocol", "SSL");\n            props.put("ssl.truststore.location", tempTruststore.toString());\n            props.put("ssl.truststore.password", truststorePassword);\n\n            KafkaProducer<String, String> producer = new KafkaProducer<>(props);\n\n            \/\/ Define the date-time formatter\n            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd 'at' HH:mm:ss");\n\n            try {\n                for (int i = 0; i < 10; i++) {\n                    String key = "Key-" + i;\n                    \n                    \/\/ Get the current timestamp\n                    String timestamp = LocalDateTime.now().format(formatter);\n                    \n                    \/\/ Include the timestamp, Kafka host, and port in the message value\n                    String value = String.format("Message-%d (Test from %s on server %s:%s)", i, timestamp, kafkaHost, kafkaPort);\n                    \n                    ProducerRecord<String, String> record = new ProducerRecord<>(topicName, key, value);\n                    producer.send(record);\n                    System.out.println("Sent message: (" + key + ", " + value + ")");\n                }\n            } catch (Exception e) {\n                e.printStackTrace();\n            } finally {\n                producer.close();\n            }\n        } catch (Exception e) {\n            e.printStackTrace();\n        }\n    }\n}\n<\/pre><\/div>\n\n\n
Consumer<\/h2>\n\n\n
\npackage com.example;\n\nimport org.apache.kafka.clients.consumer.ConsumerConfig;\nimport org.apache.kafka.clients.consumer.ConsumerRecords;\nimport org.apache.kafka.clients.consumer.KafkaConsumer;\nimport org.apache.kafka.clients.consumer.ConsumerRecord;\nimport org.apache.kafka.common.serialization.StringDeserializer;\n\nimport java.io.InputStream;\nimport java.time.Duration;\nimport java.util.Collections;\nimport java.util.Properties;\n\npublic class SimpleKafkaConsumer {\n\n    public static void main(String[] args) {\n        if (args.length < 2) {\n            System.err.println("Please provide Kafka server and port. Usage: SimpleKafkaConsumer <kafka-server> <port>");\n            System.exit(1);\n        }\n\n        String kafkaServer = args[0] + ":" + args[1];\n        String topicName = "LJRJavaTest";\n        String groupId = "test-consumer-group";\n        String truststorePassword = "truststore-password";\n\n        Properties props = new Properties();\n        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);\n        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);\n        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());\n        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());\n        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");\n\n        \/\/ Load truststore from resources\n        try (InputStream truststoreStream = SimpleKafkaConsumer.class.getResourceAsStream("\/kafka.client.truststore.jks")) {\n            if (truststoreStream == null) {\n                throw new RuntimeException("Truststore not found in resources");\n            }\n\n            \/\/ Create a temporary file to hold the truststore\n            java.nio.file.Path tempTruststore = java.nio.file.Files.createTempFile("kafka.client.truststore", ".jks");\n            java.nio.file.Files.copy(truststoreStream, tempTruststore, java.nio.file.StandardCopyOption.REPLACE_EXISTING);\n\n            props.put("security.protocol", "SSL");\n            props.put("ssl.truststore.location", tempTruststore.toString());\n            props.put("ssl.truststore.password", truststorePassword);\n\n            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);\n            consumer.subscribe(Collections.singletonList(topicName));\n\n            try {\n                while (true) {\n                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));\n                    for (ConsumerRecord<String, String> record : records) {\n                        System.out.printf("Consumed message: (key: %s, value: %s, offset: %d, partition: %d)%n",\n                                record.key(), record.value(), record.offset(), record.partition());\n                    }\n                }\n            } catch (Exception e) {\n                e.printStackTrace();\n            } finally {\n                consumer.close();\n            }\n        } catch (Exception e) {\n            e.printStackTrace();\n        }\n    }\n}\n<\/pre><\/div>\n\n\n
Keystore for Trusts<\/h2>\n\n\n\n
The cert on my Kafka server is signed by a CA that has both a root and intermediary signing certificate. To trust the chain, I need to populate rootCA.crt and intermediateCA.crt files with the correct base64 encoded public key and import both files into a new trust store:<\/p>\n\n\n
\nkeytool -importcert -file rootCA.crt -keystore kafka.client.truststore.jks -alias rootCA -storepass truststore-password\n\nkeytool -importcert -file intermediateCA.crt -keystore kafka.client.truststore.jks -alias intermediateCA -storepass truststore-password\n<\/pre><\/div>\n\n\n
\"\"<\/a><\/figure>\n\n\n\n
Then list certs in the trust store to verify the import was successful:<\/p>\n\n\n
\nkeytool -list -keystore kafka.client.truststore.jks -storepass truststore-password\n<\/pre><\/div>\n\n\n
\"\"<\/a><\/figure>\n\n\n\n
And place the jks keystore with the CA certs in project:<\/p>\n\n\n
\nC:.\n\u2502   pom.xml\n\u2502\n\u251c\u2500\u2500\u2500src\n\u2502   \u2514\u2500\u2500\u2500main\n\u2502       \u251c\u2500\u2500\u2500java\n\u2502       \u2502   \u2514\u2500\u2500\u2500com\n\u2502       \u2502       \u2514\u2500\u2500\u2500example\n\u2502       \u2502               SimpleKafkaProducer.java\n\u2502       \u2502\n\u2502       \u2514\u2500\u2500\u2500resources\n\u2502               kafka.client.truststore.jks\n\n<\/pre><\/div>\n\n\n
Build and Run<\/h2>\n\n\n\n
Create a pom.xml for the build<\/p>\n\n\n
\n<project xmlns="http:\/\/maven.apache.org\/POM\/4.0.0"\n         xmlns:xsi="http:\/\/www.w3.org\/2001\/XMLSchema-instance"\n         xsi:schemaLocation="http:\/\/maven.apache.org\/POM\/4.0.0 http:\/\/maven.apache.org\/xsd\/maven-4.0.0.xsd">\n    <modelVersion>4.0.0<\/modelVersion>\n\n    <groupId>com.example<\/groupId>\n    <artifactId>kafka-producer-example<\/artifactId>\n    <version>1.0-SNAPSHOT<\/version>\n    \n    <properties>\n        <maven.compiler.source>1.8<\/maven.compiler.source>\n        <maven.compiler.target>1.8<\/maven.compiler.target>\n    <\/properties>\n\n    <dependencies>\n        <!-- Kafka client dependency -->\n        <dependency>\n            <groupId>org.apache.kafka<\/groupId>\n            <artifactId>kafka-clients<\/artifactId>\n            <version>3.0.0<\/version>\n        <\/dependency>\n    <\/dependencies>\n\n    <build>\n        <plugins>\n            <!-- Compiler plugin to ensure Java 8 compatibility -->\n            <plugin>\n                <groupId>org.apache.maven.plugins<\/groupId>\n                <artifactId>maven-compiler-plugin<\/artifactId>\n                <version>3.8.1<\/version>\n                <configuration>\n                    <source>1.8<\/source>\n                    <target>1.8<\/target>\n                <\/configuration>\n            <\/plugin>\n            <!-- Exec Maven Plugin to run the Java class -->\n            <plugin>\n                <groupId>org.codehaus.mojo<\/groupId>\n                <artifactId>exec-maven-plugin<\/artifactId>\n                <version>3.0.0<\/version>\n                <configuration>\n                    <mainClass>com.example.SimpleKafkaProducer<\/mainClass>\n                <\/configuration>\n            <\/plugin>\n        <\/plugins>\n    <\/build>\n<\/project>\n\n<\/pre><\/div>\n\n\n
Build using `mvn clean package` <\/p>\n\n\n\n
Run the producer using ` mvn exec:java -Dexec.mainClass=”com.example.SimpleKafkaProducer” -Dexec.args=”kafkahost.example.net 9095″`<\/p>\n\n\n\n
\"\"<\/a><\/figure>\n\n\n\n
Run the consumer using ` mvn exec:java -Dexec.mainClass=”com.example.SimpleKafkaConsumer” -Dexec.args=”kafkahost.example.net 9095″`<\/p>\n\n\n\n
\"\"<\/a><\/figure>\n","protected":false},"excerpt":{"rendered":"
Producer Consumer Keystore for Trusts The cert on my Kafka server is signed by a CA that has both a root and intermediary signing certificate. To trust the chain, I need to populate rootCA.crt and intermediateCA.crt files with the correct base64 encoded public key and import both files into a new trust store: Then list …<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[33,1547,1762],"tags":[578,1361,2055,2056],"class_list":["post-11208","post","type-post","status-publish","format-standard","hentry","category-coding","category-java","category-kafka","tag-java","tag-kafka","tag-kafka-consumer","tag-kafka-producer"],"_links":{"self":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts\/11208","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=11208"}],"version-history":[{"count":1,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts\/11208\/revisions"}],"predecessor-version":[{"id":11213,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts\/11208\/revisions\/11213"}],"wp:attachment":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=11208"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=11208"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=11208"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}