Table Of Contents<\/h3>\n\n- 1. Introduction<\/a><\/dt>\n
- 2. What is Apache Kafka?<\/a><\/dt>\n
- 3. Apache Kafka Terminologies<\/a><\/dt>\n
- 4. Installing Kafka and Making a Topic<\/a><\/dt>\n
- 5. Making the Spring Boot Project with Maven<\/a><\/dt>\n
- 6. Adding Maven dependencies<\/a><\/dt>\n
- 7. Project Structure<\/a><\/dt>\n
- 8. Adding configuration<\/a><\/dt>\n
- 9. Defining the Producer configuration<\/a><\/dt>\n
- 10. Defining the Consumer configuration<\/a><\/dt>\n
- 11. Defining the Spring Boot class<\/a><\/dt>\n
- 12. Running the project<\/a><\/dt>\n
- 13. Kafka Partitioning<\/a><\/dt>\n
- 14. Conclusion<\/a><\/dt>\n
- 15. Download the Source Code<\/a><\/dt>\n<\/dl>\n<\/div>\n
2. What is Apache Kafka?<\/h2>\n
Kafka was developed at LinkedIn in the year 2010 and it was donated to Apache Software Foundation where it became one of the top projects by 2012. Apache Kafka has three main components in its ecosystem:<\/p>\n
\n- Publisher-Subscriber<\/strong>: This component of Kafka is responsible to send and consume data across Kafka Nodes (more on Nodes in coming sections) and the consumer apps which can scale with very high throughput<\/li>\n
- Kafka Streams<\/strong>: With Kafka streaming API, it is possible to process incoming data into kafka at near real-time<\/li>\n
- Connect API<\/strong>: With the Connect API, it is possible to integrate many external data sources and data sinks with Kafka<\/li>\n<\/ol>\n
For a high-level definition, we can provide a simple definition for Apache Kafka:<\/p>\n
\nApache Kafka is a distributed, fault-tolerant, horizontally-scalable, commit log.<\/p>\n<\/blockquote>\n
Let us elaborate what we just said:<\/p>\n
\n- Distributed<\/em><\/strong>: Kafka is a distributed system where all the messages are replicated across various nodes so that each server is capable of responding to the client for the messages it contains. Also, even if one node fails, other nodes can quickly take over without any downtime<\/li>\n
- Fault-tolerant<\/em><\/strong>: As Kafka does not have a Single Point of Failure<\/strong>, even if one of the node goes down, the end-user will hardly notice this as the other parts take responsibility of the messages which are lost due to the failed node<\/li>\n
- Horizontally-scalable<\/em><\/strong>: Kafka allows us to add more machines to the cluster with zero downtime. This means that if we start to face lag in messages due to a low number of servers in the cluster, we can quickly add more servers and maintain the performance of the system<\/li>\n
- Commit Log<\/em><\/strong>: A commit log refers to a structure similar to a Linked List. The order of insertion of messages is maintained and data cannot be removed from this log until a threshold time is reached<\/li>\n<\/ul>\n
A lot more concepts about Apache Kafka will become clear in the coming sections where we talk about basic terminologies used in\u00a0Apache Kafka lesson.<\/p>\n
3. Apache Kafka Terminologies<\/h2>\n
Before we can move ahead with Kafka concepts and sample project, we must understand basic terms related to Apache Kafka. Some of these are:<\/p>\n
\n- Producer<\/em><\/strong>: This component publish messages to the Kafka cluster<\/li>\n
- Consumer<\/em><\/strong>: This component consume messages from the Kafka cluster<\/li>\n
- Message<\/em><\/strong>: This is the data which is sent by the producer to the cluster<\/li>\n
- Connection<\/em><\/strong>: The producer needs to establesh a TCP connection to publish message. Same is needed in consumer app to consume data from the Kafka cluster<\/li>\n
- Topic<\/em><\/strong>: A Topic is a logical grouping of similar messages. The producer app can publish message to a particular Topic and can consume from a specific Topic<\/li>\n
- Topic Partition<\/em><\/strong>: To scale a topic memory as it can contain a lot of messages, a single Topic is divided into partitions and each partition can live on any node in the cluster, following image shows how message is written into multiple partitions:\n
![\"Spring](\"http:\/\/www.javacodegeeks.com\/wp-content\/uploads\/2018\/05\/kafka-topic-partition.png\")
<\/a>Topic partitioning in Kafka<\/figcaption><\/figure><\/li>\n- Replicas<\/em><\/strong>: As we saw in above image about topic partitioning, each message is replicated across various nodes to maintain the order and to prevent data loss when one of the node dies<\/li>\n
- Consumer Groups<\/em><\/strong>: Multiple consumers who are interested in a single topic can be kept in a group which s referred to as Consumer Groups<\/li>\n
- Offset<\/em><\/strong>: Kafka stores no information<\/strong> about which consumer is going to read what data. Each consumer keeps an offset<\/strong> value with themselves about what was the last message they read. This means that different consumers can read different messages at the same time<\/li>\n
- Node<\/em><\/strong>: A Node is just a single server in the cluster. We can add virtually any number of nodes in a cluster as we choose<\/li>\n
- Cluster<\/em><\/strong>: A group of nodes is termed as a cluster.<\/li>\n<\/ul>\n
4. Installing Kafka and Making a Topic<\/h2>\n
To download and install Kafka, we can refer to the Official Kafka guide provided here<\/a>. When the Kafka server is up and running, we can create a new topic, named as javacodegeeks<\/code> with the following command:<\/p>\nCreating a topic<\/em><\/span><\/p>\nbin\/kafka-topics --create \\\r\n --zookeeper localhost:2181 \\\r\n --replication-factor 1 --partitions 1 \\\r\n --topic javacodegeeks\r\n<\/pre>\nWe will see the following output once this command executes:<\/p>\n
![\"Spring](\"http:\/\/www.javacodegeeks.com\/wp-content\/uploads\/2018\/05\/kafka-create-topic.png\")
<\/a>Kafka create topic<\/figcaption><\/figure> <\/div> <\/div><\/p>\n5. Making the Spring Boot Project with Maven<\/h2>\n
We will be using one of the many Maven archetypes to create a sample project for our example. To create the project execute the following command in a directory that you will use as workspace:<\/p>\n
Creating a Project<\/em><\/span><\/p>\nmvn archetype:generate -DgroupId=com.javacodegeeks.example -DartifactId=JCG-BootKafka-Example -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false\r\n<\/pre>\nIf you are running maven for the first time, it will take a few seconds to accomplish the generate command because maven has to download all the required plugins and artifacts in order to make the generation task. Once we run this project, we will see following output and the project will be created:<\/p>\n
![\"Spring](\"http:\/\/www.javacodegeeks.com\/wp-content\/uploads\/2018\/05\/project-creation.png\")
<\/a>Creating Kafka project<\/figcaption><\/figure><\/p>\n6. Adding Maven dependencies<\/h2>\n
Once you have created the project, feel free to open it in your favourite IDE. Next step is to add appropriate Maven Dependencies to the project. We will work with following dependencies in our project:<\/p>\n
\nspring-boot-starter-web<\/code>: This dependency marks this project as a Web project and it adds dependencies in order to create controllers and make Web-related classes.<\/li>\nspring-kafka<\/code>: This is the dependency which brings all the Kafka related dependencies into the project classpath<\/li>\nspring-boot-starter-test<\/code>: This dependency collects all test related JARs into the project like JUnit<\/a> and Mockito<\/a>.<\/li>\n<\/ul>\nHere is the pom.xml<\/code> file with the appropriate dependencies added:<\/p>\npom.xml<\/em><\/span><\/p>\n<groupId>com.javacodegeeks.example<\/groupId>\r\n<artifactId>JCG-BootKafka-Example<\/artifactId>\r\n<packaging>jar<\/packaging>\r\n<version>1.0-SNAPSHOT<\/version>\r\n<name>JCG-BootKafka-Example<\/name>\r\n\r\n<parent>\r\n <groupId>org.springframework.boot<\/groupId>\r\n <artifactId>spring-boot-starter-parent<\/artifactId>\r\n <version>1.5.10.RELEASE<\/version>\r\n <relativePath\/> <!-- lookup parent from repository -->\r\n<\/parent>\r\n\r\n<properties>\r\n <project.build.sourceEncoding>UTF-8<\/project.build.sourceEncoding>\r\n <project.reporting.outputEncoding>UTF-8<\/project.reporting.outputEncoding>\r\n <java.version>1.8<\/java.version>\r\n<\/properties>\r\n\r\n<dependencies>\r\n\r\n <dependency>\r\n <groupId>org.springframework.boot<\/groupId>\r\n <artifactId>spring-boot-starter-web<\/artifactId>\r\n <\/dependency>\r\n\r\n <dependency>\r\n <groupId>org.springframework.kafka<\/groupId>\r\n <artifactId>spring-kafka<\/artifactId>\r\n <version>1.1.3.RELEASE<\/version>\r\n <\/dependency>\r\n\r\n <dependency>\r\n <groupId>org.springframework.boot<\/groupId>\r\n <artifactId>spring-boot-starter-test<\/artifactId>\r\n <scope>test<\/scope>\r\n <\/dependency>\r\n\r\n<\/dependencies>\r\n\r\n<build>\r\n <plugins>\r\n <plugin>\r\n <groupId>org.springframework.boot<\/groupId>\r\n <artifactId>spring-boot-maven-plugin<\/artifactId>\r\n <\/plugin>\r\n <\/plugins>\r\n<\/build>\r\n<\/pre>\nFind the latest Maven dependencies on Maven Central<\/a>.<\/p>\nFinally, to understand all the JARs which are added to the project when we added this dependency, we can run a simple Maven command which allows us to see a complete Dependency Tree for a project when we add some dependencies to it. Here is a command which we can use:<\/p>\n
Check Dependency Tree<\/em><\/span><\/p>\nmvn dependency:tree\r\n<\/pre>\nWhen we run this command, it will show us the following Dependency Tree:<\/p>\n
![\"Spring](\"http:\/\/www.javacodegeeks.com\/wp-content\/uploads\/2018\/05\/kafka-dependency.png\")
<\/a>Checking Dependency<\/figcaption><\/figure><\/p>\nNoticed something? So many dependencies were added by just adding four dependencies to the project. Spring Boot collects all related dependencies itself and leave nothing for us in that matter. The biggest advantage is that all these dependencies are guranteed to be copatible with each other<\/strong>.<\/p>\n7. Project Structure<\/h2>\n
Before we move on and start working on the code for the project, let\u2019s present here the project structure we will have once we\u2019re finished adding all the code to the project:<\/p>\n
![\"Spring](\"http:\/\/www.javacodegeeks.com\/wp-content\/uploads\/2018\/05\/project-structure.png\")
<\/a>Project Structure<\/figcaption><\/figure><\/p>\nWe have divided the project into multiple packages so that the principle of separation of concern<\/a> is followed and code remains modular.<\/p>\n8. Adding configuration<\/h2>\n
Before we can start writing code for our project, we need to provide some properties in the application.properties<\/code> file of our Spring Boot project:<\/p>\napplication.properties<\/em><\/span><\/p>\n#Kafka Topic\r\nmessage.topic.name=javacodegeeks\r\n\r\nspring.kafka.bootstrap-servers=localhost:9092\r\n\r\n#Unique String which identifies which consumer group this consumer belongs to\r\nspring.kafka.consumer.group-id=jcg-group\r\n<\/pre>\nThese are some properties which we will be using in our project as the topic and group ID for the message we will publish and consume. Also, 9092<\/code> is the default port for Apache Kafka. Note that we could have defined multiple topics here as well with different name for the keys.<\/p>\n9. Defining the Producer configuration<\/h2>\n
We will start by defining the configuration for a Producer. The only property we need to define mandatorily for a Kafka Producer is the address for Kafka server with its port.<\/p>\n
KafkaProducerConfig.java<\/em><\/span><\/p>\npackage com.javacodegeeks.example.config;\r\n\r\nimport org.apache.kafka.clients.producer.ProducerConfig;\r\nimport org.apache.kafka.common.serialization.StringSerializer;\r\nimport org.springframework.beans.factory.annotation.Value;\r\nimport org.springframework.context.annotation.Bean;\r\nimport org.springframework.context.annotation.Configuration;\r\nimport org.springframework.kafka.core.DefaultKafkaProducerFactory;\r\nimport org.springframework.kafka.core.KafkaTemplate;\r\nimport org.springframework.kafka.core.ProducerFactory;\r\n\r\nimport java.util.HashMap;\r\nimport java.util.Map;\r\n\r\n@Configuration\r\npublic class KafkaProducerConfig {\r\n\r\n @Value(\"${spring.kafka.bootstrap-servers}\")\r\n private String bootstrapAddress;\r\n\r\n @Bean\r\n public ProducerFactory<String, String> producerFactory() {\r\n Map<String, Object> configProps = new HashMap<>();\r\n configProps.put(\r\n ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);\r\n configProps.put(\r\n ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);\r\n configProps.put(\r\n ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);\r\n return new DefaultKafkaProducerFactory<>(configProps);\r\n }\r\n\r\n @Bean\r\n public KafkaTemplate<String, String> kafkaTemplate() {\r\n return new KafkaTemplate<>(producerFactory());\r\n }\r\n}\r\n<\/pre>\nAlthough above class definition is pretty simple, we still have some points which we need to understand:<\/p>\n
\n@Configuration<\/code>: This class is defined as a configuration class which means that this class will be picked up by Spring Boot automatically and all beans defined inside this class will be managed by Spring container automatically.<\/li>\n- We defined a bean for
ProducerFactory<\/code> which takes input as various properties like the Kafka server address and other serialization properties which hels to code and decode the message sent through the Kafka producer bean.<\/li>\n- Finally, we defined a bean for
KafkaTemplate<\/code> which is the actual API object which will be used to publish the message on a Kafka topic.<\/li>\n<\/ul>\n10. Defining the Consumer configuration<\/h2>\n
As we are making Kafka producer and consumer in the same app for demonstration purposes, we will also define a consumer config class which will just contain basic properties for a Kafka consumer. This class can be put inside any project which isn’t a producer and only a Kafka consumer as well without any changes. Let’s look at the configuration definition:<\/p>\n
KafkaConsumerConfig.java<\/em><\/span><\/p>\npackage com.javacodegeeks.example.config;\r\n\r\nimport org.apache.kafka.clients.consumer.ConsumerConfig;\r\nimport org.apache.kafka.common.serialization.StringDeserializer;\r\nimport org.springframework.beans.factory.annotation.Value;\r\nimport org.springframework.context.annotation.Bean;\r\nimport org.springframework.context.annotation.Configuration;\r\nimport org.springframework.kafka.annotation.EnableKafka;\r\nimport org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;\r\nimport org.springframework.kafka.core.ConsumerFactory;\r\nimport org.springframework.kafka.core.DefaultKafkaConsumerFactory;\r\n\r\nimport java.util.HashMap;\r\nimport java.util.Map;\r\n\r\n@EnableKafka\r\n@Configuration\r\npublic class KafkaConsumerConfig {\r\n\r\n @Value(\"${spring.kafka.bootstrap-servers}\")\r\n private String bootstrapAddress;\r\n\r\n @Value(\"${spring.kafka.consumer.group-id}\")\r\n private String groupId;\r\n\r\n @Bean\r\n public ConsumerFactory<String, String> consumerFactory() {\r\n Map<String, Object> props = new HashMap<>();\r\n props.put(\r\n ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);\r\n props.put(\r\n ConsumerConfig.GROUP_ID_CONFIG, groupId);\r\n props.put(\r\n ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);\r\n props.put(\r\n ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);\r\n return new DefaultKafkaConsumerFactory<>(props);\r\n }\r\n\r\n @Bean\r\n public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {\r\n ConcurrentKafkaListenerContainerFactory<String, String> factory\r\n = new ConcurrentKafkaListenerContainerFactory<>();\r\n factory.setConsumerFactory(consumerFactory());\r\n return factory;\r\n }\r\n}\r\n<\/pre>\nThe configuration we provided is very similar to the producer configuration. The only difference to note here is:<\/p>\n
\n- We defined a
ConsumerFactory<\/code> class object bean which considers the Kafka server address and the consumer group ID as well which this Kafka consumer app belongs to. We have provided a unique String to the consumer as only a unique string is acceptable<\/li>\n- Finally, we defined a
ConcurrentKafkaListenerContainerFactory<\/code> which makes sure that this consumer application can consume Kafka messages at a concurrent pace and can give a consistent throughput even when published messages are high in number.<\/li>\n<\/ul>\n11. Defining the Spring Boot class<\/h2>\n
At the final stage, we will make the Spring Boot class with which we can publish a message, consume the message on the same topic. Here is the class definition for the main class:<\/p>\n
KafkaApp.java<\/em><\/span><\/p>\npackage com.javacodegeeks.example;\r\n\r\nimport org.slf4j.Logger;\r\nimport org.slf4j.LoggerFactory;\r\nimport org.springframework.beans.factory.annotation.Autowired;\r\nimport org.springframework.beans.factory.annotation.Value;\r\nimport org.springframework.boot.CommandLineRunner;\r\nimport org.springframework.boot.SpringApplication;\r\nimport org.springframework.boot.autoconfigure.SpringBootApplication;\r\nimport org.springframework.kafka.annotation.KafkaListener;\r\nimport org.springframework.kafka.core.KafkaTemplate;\r\n\r\n@SpringBootApplication\r\npublic class KafkaApp implements CommandLineRunner {\r\n\r\n private static final Logger LOG = LoggerFactory.getLogger(\"KafkaApp\");\r\n\r\n @Value(\"${message.topic.name}\")\r\n private String topicName;\r\n\r\n private final KafkaTemplate<String, String> kafkaTemplate;\r\n\r\n @Autowired\r\n public KafkaApp(KafkaTemplate<String, String> kafkaTemplate) {\r\n this.kafkaTemplate = kafkaTemplate;\r\n }\r\n\r\n public static void main(String[] args) {\r\n SpringApplication.run(KafkaApp.class, args);\r\n }\r\n\r\n @Override\r\n public void run(String... strings) {\r\n kafkaTemplate.send(topicName, \"Hello Geek!\");\r\n LOG.info(\"Published message to topic: {}.\", topicName);\r\n }\r\n\r\n @KafkaListener(topics = \"javacodegeeks\", group = \"jcg-group\")\r\n public void listen(String message) {\r\n LOG.info(\"Received message in JCG group: {}\", message);\r\n }\r\n\r\n}\r\n<\/pre>\nWe have used an interface CommandLineRunner<\/code> to make this class run code with which we can test the producer and config class code we wrote. In this class, we publish a message to the specified topic and listen for it in the listener method we defined in the same class.<\/p>\nIn the next section, we will run our project with a simple Maven command.<\/p>\n
12. Running the project<\/h2>\n
Now that the main class definition is done, we can run our project. Running the application is easy with maven, just use the following command:<\/p>\n
Running the Project<\/em><\/span><\/p>\nmvn spring-boot:run\r\n<\/pre>\nOnce we execute the above command, we will see that a message has been published on the specified topic and same app consumed the message as well:<\/p>\n
2. What is Apache Kafka?<\/h2>\n
Kafka was developed at LinkedIn in the year 2010 and it was donated to Apache Software Foundation where it became one of the top projects by 2012. Apache Kafka has three main components in its ecosystem:<\/p>\n
- \n
- Publisher-Subscriber<\/strong>: This component of Kafka is responsible to send and consume data across Kafka Nodes (more on Nodes in coming sections) and the consumer apps which can scale with very high throughput<\/li>\n
- Kafka Streams<\/strong>: With Kafka streaming API, it is possible to process incoming data into kafka at near real-time<\/li>\n
- Connect API<\/strong>: With the Connect API, it is possible to integrate many external data sources and data sinks with Kafka<\/li>\n<\/ol>\n
For a high-level definition, we can provide a simple definition for Apache Kafka:<\/p>\n
\n
Apache Kafka is a distributed, fault-tolerant, horizontally-scalable, commit log.<\/p>\n<\/blockquote>\n
Let us elaborate what we just said:<\/p>\n
- \n
- Distributed<\/em><\/strong>: Kafka is a distributed system where all the messages are replicated across various nodes so that each server is capable of responding to the client for the messages it contains. Also, even if one node fails, other nodes can quickly take over without any downtime<\/li>\n
- Fault-tolerant<\/em><\/strong>: As Kafka does not have a Single Point of Failure<\/strong>, even if one of the node goes down, the end-user will hardly notice this as the other parts take responsibility of the messages which are lost due to the failed node<\/li>\n
- Horizontally-scalable<\/em><\/strong>: Kafka allows us to add more machines to the cluster with zero downtime. This means that if we start to face lag in messages due to a low number of servers in the cluster, we can quickly add more servers and maintain the performance of the system<\/li>\n
- Commit Log<\/em><\/strong>: A commit log refers to a structure similar to a Linked List. The order of insertion of messages is maintained and data cannot be removed from this log until a threshold time is reached<\/li>\n<\/ul>\n
A lot more concepts about Apache Kafka will become clear in the coming sections where we talk about basic terminologies used in\u00a0Apache Kafka lesson.<\/p>\n
3. Apache Kafka Terminologies<\/h2>\n
Before we can move ahead with Kafka concepts and sample project, we must understand basic terms related to Apache Kafka. Some of these are:<\/p>\n
- \n
- Producer<\/em><\/strong>: This component publish messages to the Kafka cluster<\/li>\n
- Consumer<\/em><\/strong>: This component consume messages from the Kafka cluster<\/li>\n
- Message<\/em><\/strong>: This is the data which is sent by the producer to the cluster<\/li>\n
- Connection<\/em><\/strong>: The producer needs to establesh a TCP connection to publish message. Same is needed in consumer app to consume data from the Kafka cluster<\/li>\n
- Topic<\/em><\/strong>: A Topic is a logical grouping of similar messages. The producer app can publish message to a particular Topic and can consume from a specific Topic<\/li>\n
- Topic Partition<\/em><\/strong>: To scale a topic memory as it can contain a lot of messages, a single Topic is divided into partitions and each partition can live on any node in the cluster, following image shows how message is written into multiple partitions:\n
<\/a>Topic partitioning in Kafka<\/figcaption><\/figure><\/li>\n - Replicas<\/em><\/strong>: As we saw in above image about topic partitioning, each message is replicated across various nodes to maintain the order and to prevent data loss when one of the node dies<\/li>\n
- Consumer Groups<\/em><\/strong>: Multiple consumers who are interested in a single topic can be kept in a group which s referred to as Consumer Groups<\/li>\n
- Offset<\/em><\/strong>: Kafka stores no information<\/strong> about which consumer is going to read what data. Each consumer keeps an offset<\/strong> value with themselves about what was the last message they read. This means that different consumers can read different messages at the same time<\/li>\n
- Node<\/em><\/strong>: A Node is just a single server in the cluster. We can add virtually any number of nodes in a cluster as we choose<\/li>\n
- Cluster<\/em><\/strong>: A group of nodes is termed as a cluster.<\/li>\n<\/ul>\n
4. Installing Kafka and Making a Topic<\/h2>\n
To download and install Kafka, we can refer to the Official Kafka guide provided here<\/a>. When the Kafka server is up and running, we can create a new topic, named as
javacodegeeks<\/code> with the following command:<\/p>\nCreating a topic<\/em><\/span><\/p>\n
bin\/kafka-topics --create \\\r\n --zookeeper localhost:2181 \\\r\n --replication-factor 1 --partitions 1 \\\r\n --topic javacodegeeks\r\n<\/pre>\n
We will see the following output once this command executes:<\/p>\n
<\/a>Kafka create topic<\/figcaption><\/figure> <\/div> <\/div><\/p>\n5. Making the Spring Boot Project with Maven<\/h2>\n
We will be using one of the many Maven archetypes to create a sample project for our example. To create the project execute the following command in a directory that you will use as workspace:<\/p>\n
Creating a Project<\/em><\/span><\/p>\n
mvn archetype:generate -DgroupId=com.javacodegeeks.example -DartifactId=JCG-BootKafka-Example -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false\r\n<\/pre>\n
If you are running maven for the first time, it will take a few seconds to accomplish the generate command because maven has to download all the required plugins and artifacts in order to make the generation task. Once we run this project, we will see following output and the project will be created:<\/p>\n
<\/a>Creating Kafka project<\/figcaption><\/figure><\/p>\n 6. Adding Maven dependencies<\/h2>\n
Once you have created the project, feel free to open it in your favourite IDE. Next step is to add appropriate Maven Dependencies to the project. We will work with following dependencies in our project:<\/p>\n
- \n
spring-boot-starter-web<\/code>: This dependency marks this project as a Web project and it adds dependencies in order to create controllers and make Web-related classes.<\/li>\nspring-kafka<\/code>: This is the dependency which brings all the Kafka related dependencies into the project classpath<\/li>\nspring-boot-starter-test<\/code>: This dependency collects all test related JARs into the project like JUnit<\/a> and Mockito<\/a>.<\/li>\n<\/ul>\nHere is the
pom.xml<\/code> file with the appropriate dependencies added:<\/p>\npom.xml<\/em><\/span><\/p>\n
<groupId>com.javacodegeeks.example<\/groupId>\r\n<artifactId>JCG-BootKafka-Example<\/artifactId>\r\n<packaging>jar<\/packaging>\r\n<version>1.0-SNAPSHOT<\/version>\r\n<name>JCG-BootKafka-Example<\/name>\r\n\r\n<parent>\r\n <groupId>org.springframework.boot<\/groupId>\r\n <artifactId>spring-boot-starter-parent<\/artifactId>\r\n <version>1.5.10.RELEASE<\/version>\r\n <relativePath\/> <!-- lookup parent from repository -->\r\n<\/parent>\r\n\r\n<properties>\r\n <project.build.sourceEncoding>UTF-8<\/project.build.sourceEncoding>\r\n <project.reporting.outputEncoding>UTF-8<\/project.reporting.outputEncoding>\r\n <java.version>1.8<\/java.version>\r\n<\/properties>\r\n\r\n<dependencies>\r\n\r\n <dependency>\r\n <groupId>org.springframework.boot<\/groupId>\r\n <artifactId>spring-boot-starter-web<\/artifactId>\r\n <\/dependency>\r\n\r\n <dependency>\r\n <groupId>org.springframework.kafka<\/groupId>\r\n <artifactId>spring-kafka<\/artifactId>\r\n <version>1.1.3.RELEASE<\/version>\r\n <\/dependency>\r\n\r\n <dependency>\r\n <groupId>org.springframework.boot<\/groupId>\r\n <artifactId>spring-boot-starter-test<\/artifactId>\r\n <scope>test<\/scope>\r\n <\/dependency>\r\n\r\n<\/dependencies>\r\n\r\n<build>\r\n <plugins>\r\n <plugin>\r\n <groupId>org.springframework.boot<\/groupId>\r\n <artifactId>spring-boot-maven-plugin<\/artifactId>\r\n <\/plugin>\r\n <\/plugins>\r\n<\/build>\r\n<\/pre>\n
Find the latest Maven dependencies on Maven Central<\/a>.<\/p>\n
Finally, to understand all the JARs which are added to the project when we added this dependency, we can run a simple Maven command which allows us to see a complete Dependency Tree for a project when we add some dependencies to it. Here is a command which we can use:<\/p>\n
Check Dependency Tree<\/em><\/span><\/p>\n
mvn dependency:tree\r\n<\/pre>\n
When we run this command, it will show us the following Dependency Tree:<\/p>\n
<\/a>Checking Dependency<\/figcaption><\/figure><\/p>\n Noticed something? So many dependencies were added by just adding four dependencies to the project. Spring Boot collects all related dependencies itself and leave nothing for us in that matter. The biggest advantage is that all these dependencies are guranteed to be copatible with each other<\/strong>.<\/p>\n
7. Project Structure<\/h2>\n
Before we move on and start working on the code for the project, let\u2019s present here the project structure we will have once we\u2019re finished adding all the code to the project:<\/p>\n
<\/a>Project Structure<\/figcaption><\/figure><\/p>\n We have divided the project into multiple packages so that the principle of separation of concern<\/a> is followed and code remains modular.<\/p>\n
8. Adding configuration<\/h2>\n
Before we can start writing code for our project, we need to provide some properties in the
application.properties<\/code> file of our Spring Boot project:<\/p>\napplication.properties<\/em><\/span><\/p>\n
#Kafka Topic\r\nmessage.topic.name=javacodegeeks\r\n\r\nspring.kafka.bootstrap-servers=localhost:9092\r\n\r\n#Unique String which identifies which consumer group this consumer belongs to\r\nspring.kafka.consumer.group-id=jcg-group\r\n<\/pre>\n
These are some properties which we will be using in our project as the topic and group ID for the message we will publish and consume. Also,
9092<\/code> is the default port for Apache Kafka. Note that we could have defined multiple topics here as well with different name for the keys.<\/p>\n9. Defining the Producer configuration<\/h2>\n
We will start by defining the configuration for a Producer. The only property we need to define mandatorily for a Kafka Producer is the address for Kafka server with its port.<\/p>\n
KafkaProducerConfig.java<\/em><\/span><\/p>\n
package com.javacodegeeks.example.config;\r\n\r\nimport org.apache.kafka.clients.producer.ProducerConfig;\r\nimport org.apache.kafka.common.serialization.StringSerializer;\r\nimport org.springframework.beans.factory.annotation.Value;\r\nimport org.springframework.context.annotation.Bean;\r\nimport org.springframework.context.annotation.Configuration;\r\nimport org.springframework.kafka.core.DefaultKafkaProducerFactory;\r\nimport org.springframework.kafka.core.KafkaTemplate;\r\nimport org.springframework.kafka.core.ProducerFactory;\r\n\r\nimport java.util.HashMap;\r\nimport java.util.Map;\r\n\r\n@Configuration\r\npublic class KafkaProducerConfig {\r\n\r\n @Value(\"${spring.kafka.bootstrap-servers}\")\r\n private String bootstrapAddress;\r\n\r\n @Bean\r\n public ProducerFactory<String, String> producerFactory() {\r\n Map<String, Object> configProps = new HashMap<>();\r\n configProps.put(\r\n ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);\r\n configProps.put(\r\n ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);\r\n configProps.put(\r\n ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);\r\n return new DefaultKafkaProducerFactory<>(configProps);\r\n }\r\n\r\n @Bean\r\n public KafkaTemplate<String, String> kafkaTemplate() {\r\n return new KafkaTemplate<>(producerFactory());\r\n }\r\n}\r\n<\/pre>\nAlthough above class definition is pretty simple, we still have some points which we need to understand:<\/p>\n
- \n
@Configuration<\/code>: This class is defined as a configuration class which means that this class will be picked up by Spring Boot automatically and all beans defined inside this class will be managed by Spring container automatically.<\/li>\n- We defined a bean for
ProducerFactory<\/code> which takes input as various properties like the Kafka server address and other serialization properties which hels to code and decode the message sent through the Kafka producer bean.<\/li>\n- Finally, we defined a bean for
KafkaTemplate<\/code> which is the actual API object which will be used to publish the message on a Kafka topic.<\/li>\n<\/ul>\n10. Defining the Consumer configuration<\/h2>\n
As we are making Kafka producer and consumer in the same app for demonstration purposes, we will also define a consumer config class which will just contain basic properties for a Kafka consumer. This class can be put inside any project which isn’t a producer and only a Kafka consumer as well without any changes. Let’s look at the configuration definition:<\/p>\n
KafkaConsumerConfig.java<\/em><\/span><\/p>\n
package com.javacodegeeks.example.config;\r\n\r\nimport org.apache.kafka.clients.consumer.ConsumerConfig;\r\nimport org.apache.kafka.common.serialization.StringDeserializer;\r\nimport org.springframework.beans.factory.annotation.Value;\r\nimport org.springframework.context.annotation.Bean;\r\nimport org.springframework.context.annotation.Configuration;\r\nimport org.springframework.kafka.annotation.EnableKafka;\r\nimport org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;\r\nimport org.springframework.kafka.core.ConsumerFactory;\r\nimport org.springframework.kafka.core.DefaultKafkaConsumerFactory;\r\n\r\nimport java.util.HashMap;\r\nimport java.util.Map;\r\n\r\n@EnableKafka\r\n@Configuration\r\npublic class KafkaConsumerConfig {\r\n\r\n @Value(\"${spring.kafka.bootstrap-servers}\")\r\n private String bootstrapAddress;\r\n\r\n @Value(\"${spring.kafka.consumer.group-id}\")\r\n private String groupId;\r\n\r\n @Bean\r\n public ConsumerFactory<String, String> consumerFactory() {\r\n Map<String, Object> props = new HashMap<>();\r\n props.put(\r\n ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);\r\n props.put(\r\n ConsumerConfig.GROUP_ID_CONFIG, groupId);\r\n props.put(\r\n ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);\r\n props.put(\r\n ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);\r\n return new DefaultKafkaConsumerFactory<>(props);\r\n }\r\n\r\n @Bean\r\n public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {\r\n ConcurrentKafkaListenerContainerFactory<String, String> factory\r\n = new ConcurrentKafkaListenerContainerFactory<>();\r\n factory.setConsumerFactory(consumerFactory());\r\n return factory;\r\n }\r\n}\r\n<\/pre>\nThe configuration we provided is very similar to the producer configuration. The only difference to note here is:<\/p>\n
- \n
- We defined a
ConsumerFactory<\/code> class object bean which considers the Kafka server address and the consumer group ID as well which this Kafka consumer app belongs to. We have provided a unique String to the consumer as only a unique string is acceptable<\/li>\n- Finally, we defined a
ConcurrentKafkaListenerContainerFactory<\/code> which makes sure that this consumer application can consume Kafka messages at a concurrent pace and can give a consistent throughput even when published messages are high in number.<\/li>\n<\/ul>\n11. Defining the Spring Boot class<\/h2>\n
At the final stage, we will make the Spring Boot class with which we can publish a message, consume the message on the same topic. Here is the class definition for the main class:<\/p>\n
KafkaApp.java<\/em><\/span><\/p>\n
package com.javacodegeeks.example;\r\n\r\nimport org.slf4j.Logger;\r\nimport org.slf4j.LoggerFactory;\r\nimport org.springframework.beans.factory.annotation.Autowired;\r\nimport org.springframework.beans.factory.annotation.Value;\r\nimport org.springframework.boot.CommandLineRunner;\r\nimport org.springframework.boot.SpringApplication;\r\nimport org.springframework.boot.autoconfigure.SpringBootApplication;\r\nimport org.springframework.kafka.annotation.KafkaListener;\r\nimport org.springframework.kafka.core.KafkaTemplate;\r\n\r\n@SpringBootApplication\r\npublic class KafkaApp implements CommandLineRunner {\r\n\r\n private static final Logger LOG = LoggerFactory.getLogger(\"KafkaApp\");\r\n\r\n @Value(\"${message.topic.name}\")\r\n private String topicName;\r\n\r\n private final KafkaTemplate<String, String> kafkaTemplate;\r\n\r\n @Autowired\r\n public KafkaApp(KafkaTemplate<String, String> kafkaTemplate) {\r\n this.kafkaTemplate = kafkaTemplate;\r\n }\r\n\r\n public static void main(String[] args) {\r\n SpringApplication.run(KafkaApp.class, args);\r\n }\r\n\r\n @Override\r\n public void run(String... strings) {\r\n kafkaTemplate.send(topicName, \"Hello Geek!\");\r\n LOG.info(\"Published message to topic: {}.\", topicName);\r\n }\r\n\r\n @KafkaListener(topics = \"javacodegeeks\", group = \"jcg-group\")\r\n public void listen(String message) {\r\n LOG.info(\"Received message in JCG group: {}\", message);\r\n }\r\n\r\n}\r\n<\/pre>\nWe have used an interface
CommandLineRunner<\/code> to make this class run code with which we can test the producer and config class code we wrote. In this class, we publish a message to the specified topic and listen for it in the listener method we defined in the same class.<\/p>\nIn the next section, we will run our project with a simple Maven command.<\/p>\n
12. Running the project<\/h2>\n
Now that the main class definition is done, we can run our project. Running the application is easy with maven, just use the following command:<\/p>\n
Running the Project<\/em><\/span><\/p>\n
mvn spring-boot:run\r\n<\/pre>\n
Once we execute the above command, we will see that a message has been published on the specified topic and same app consumed the message as well:<\/p>\n
- Finally, we defined a
- We defined a bean for
- Consumer<\/em><\/strong>: This component consume messages from the Kafka cluster<\/li>\n
- Fault-tolerant<\/em><\/strong>: As Kafka does not have a Single Point of Failure<\/strong>, even if one of the node goes down, the end-user will hardly notice this as the other parts take responsibility of the messages which are lost due to the failed node<\/li>\n
- Kafka Streams<\/strong>: With Kafka streaming API, it is possible to process incoming data into kafka at near real-time<\/li>\n
![\"Spring](\"http:\/\/www.javacodegeeks.com\/wp-content\/uploads\/2018\/05\/kafka-project-running.png\")
<\/a>![\"Spring](\"http:\/\/www.javacodegeeks.com\/wp-content\/uploads\/2018\/05\/kafka-topic-partition-blocks.png\")
<\/a>