![\"spring](\"http:\/\/examples.javacodegeeks.com\/wp-content\/uploads\/2017\/12\/jcg2-usecase-final2.jpg\")
<\/a>The application is implemented in two Spring Boot projects:
a) ‘library’ which is the producer that sends Book messages to a Kafka broker
b) ‘reader’ which is the consumer that receives books.<\/p>\n
In the code, we use four-channel classes from Spring Integration: The One key point to note is that each record published to a topic is delivered to one consumer instance within each subscribing consumer group. Hence we will run two instances of the reader project, each belonging to a different consumer group and subscribing to a different combination of topics. Both the consumer group and topics set are given to the application as command-line arguments.<\/p>\n I have used the following technologies for this application:<\/p>\n library<\/b>: This is a maven-based project, so all the dependencies are specified in the pom.xml file.<\/p>\n pom.xml<\/em><\/span><\/p>\n Below is the Book.java<\/em><\/span><\/p>\n A Book has an Below is the BookPublisher.java<\/em><\/span><\/p>\n The main functionality of this class is to create and return a list of twenty books, five each with the fantasy, horror, romance, and thriller genres. There is a book creation method for each of the genre type, which call a utility method Below is the ProducerChannelConfig.java<\/em><\/span><\/p>\n Below is the Library.java<\/em><\/span><\/p>\n From the application context, a Below is the application.properties<\/em><\/span><\/p>\n Here we specify port 9092 for the Kafka server to connect to.<\/p>\n Next, we take a look at the consumer side of the application. pom.xml<\/em><\/span><\/p>\n Below is the ConsumerChannelConfig.java<\/em><\/span><\/p>\n Below is the SubscribedReader.java<\/em><\/span><\/p>\n In the main method, we first check if there are any command line arguments. If they are present and are valid topics, they are added to an In the Below is the application.properties<\/em><\/span><\/p>\n A dummy topic is specified so that when the consumer application starts, the message channel is properly configured. Next, the actual topics to listen to are added. The dummy topic is never used by the producer to send messages.<\/p>\n You will need five terminal windows. Terminal 2: Start KafkaServer. Go to your Kafka installation folder and run the following command:<\/p>\n Terminal 3: Start the first consumer with group id “group-one” and subscribed to fantasy and horror genres. Changed directory to the reader and run the following command:<\/p>\n Terminal 4: Start the second consumer with group id “group-one” and subscribed to horror, romance and thriller genres. Change directory to the reader and run the following command:<\/p>\n Terminal 5: Run producer. In the library folder, run the following command:<\/p>\n You will see the received messages in terminals 3 and 4. Please note that you can run the commands in terminals 3, 4, and 5 in any order. Due to Kafka’s retention time policy, defaulted to 7 days and its file-like persistence mechanics, you will still get the same output.<\/p>\n In this example, we have seen the publish-subscribe mechanism provided by Apache Kafka and the methods by which Spring Integration enables applications to connect with it. We have also touched upon different message channels available with Spring Integration and described their key features.<\/p>\n The following resources will be very useful to get additional information and insights on concepts discussed in this article:<\/p>\nMessageChannel<\/code>, DirectChannel<\/code>, PollableChannel<\/code>, and QueueChannel<\/code>. MessageChannel<\/code> is an interface that is implemented by all Spring Integration channels. It declares the send method which the concrete classes define how a sender sends a message to the channel.<\/p>\nDirectChannel<\/code> implements the SubscribableChannel<\/code> (which extends MessageChannel<\/code>) and has point-to-point semantics, that is, it will only send each Message<\/code> to a single subscriber.<\/p>\nPollableChannel<\/code> is an interface that extends the MessageChannel<\/code> and is used for receiving messages. Classes implementing this interface provide functionality for polling messages from a channel.<\/p>\nQueueChannel<\/code> implements multiple interfaces. It wraps a queue, provides point-to-point semantics, and has the functionality to filter and purge messages that satisfy certain criteria.<\/p>\n3. Environment<\/h2>\n
\n
4. Source Code<\/h2>\n
<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<project xmlns=\"http:\/\/maven.apache.org\/POM\/4.0.0\" xmlns:xsi=\"http:\/\/www.w3.org\/2001\/XMLSchema-instance\"\nxsi: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>org.javacodegeeks.springintegration.kafka<\/groupId>\n <artifactId>producer<\/artifactId>\n <version>0.0.1-SNAPSHOT<\/version>\n <packaging>jar<\/packaging>\n\n <name>producer<\/name>\n <description>Kafka producer with Spring Boot<\/description>\n\n <parent>\n <groupId>org.springframework.boot<\/groupId>\n <artifactId>spring-boot-starter-parent<\/artifactId>\n <version>1.5.9.RELEASE<\/version>\n <relativePath \/> <!-- lookup parent from repository -->\n <\/parent>\n\n <properties>\n <project.build.sourceEncoding>UTF-8<\/project.build.sourceEncoding>\n <project.reporting.outputEncoding>UTF-8<\/project.reporting.outputEncoding>\n <java.version>1.8<\/java.version>\n <\/properties>\n\n <dependencies>\n <dependency>\n <groupId>org.apache.kafka<\/groupId>\n <artifactId>kafka-clients<\/artifactId>\n <version>1.0.0<\/version>\n <\/dependency>\n <dependency>\n <groupId>org.springframework.boot<\/groupId>\n <artifactId>spring-boot-starter-integration<\/artifactId>\n <\/dependency>\n <dependency>\n <groupId>org.springframework.integration<\/groupId>\n <artifactId>spring-integration-kafka<\/artifactId>\n <version>2.3.0.RELEASE<\/version>\n <\/dependency>\n <dependency>\n <groupId>org.springframework.kafka<\/groupId>\n <artifactId>spring-kafka<\/artifactId>\n <version>1.3.2.RELEASE<\/version>\n <\/dependency>\n <dependency>\n <groupId>org.projectlombok<\/groupId>\n <artifactId>lombok<\/artifactId>\n <optional>true<\/optional>\n <\/dependency>\n <dependency>\n <groupId>org.springframework.boot<\/groupId>\n <artifactId>spring-boot-starter-test<\/artifactId>\n <scope>test<\/scope>\n <\/dependency>\n <\/dependencies>\n <build>\n <plugins>\n <plugin>\n <groupId>org.springframework.boot<\/groupId>\n <artifactId>spring-boot-maven-plugin<\/artifactId>\n <\/plugin>\n <\/plugins>\n <\/build>\n<\/project>\n\n<\/pre>\n
Book<\/code> class that serves as the model for the application.<\/p>\npackage org.javacodegeeks.springintegration.kafka.model;\n\nimport lombok.Getter;\nimport lombok.NoArgsConstructor;\nimport lombok.Setter;\nimport lombok.ToString;\n\n@Getter\n@Setter\n@NoArgsConstructor\n@ToString\npublic class Book {\n\n\tpublic enum Genre {\n\t\tfantasy, horror, romance, thriller\n\t}\n\n\tprivate long bookId;\n\tprivate String title;\n\tprivate Genre genre;\n}\n<\/pre>\nenum<\/code> indicating which genre it belongs to. The other two properties are bookId<\/code> and title<\/code>. The lombok<\/code> annotations inject the setters, getters, a no-argument constructor and the toString()<\/code> method to all the members.<\/p>\nBookPublisher<\/code> class that initiates the message flow in the application.<\/p>\npackage org.javacodegeeks.springintegration.kafka.incoming;\n\nimport java.util.ArrayList;\nimport java.util.List;\n\nimport org.javacodegeeks.springintegration.kafka.model.Book;\nimport org.javacodegeeks.springintegration.kafka.model.Book.Genre;\nimport org.springframework.stereotype.Component;\n\n@Component\npublic class BookPublisher {\n\tprivate long nextBookId;\n\n\tpublic BookPublisher() {\n\t\tthis.nextBookId = 1001l;\n\t}\n\n\tpublic List getBooks() {\n\t\tList books = new ArrayList();\n\n\t\tbooks.add(createFantasyBook());\n\t\tbooks.add(createFantasyBook());\n\t\tbooks.add(createFantasyBook());\n\t\tbooks.add(createFantasyBook());\n\t\tbooks.add(createFantasyBook());\n\t\tbooks.add(createHorrorBook());\n\t\tbooks.add(createHorrorBook());\n\t\tbooks.add(createHorrorBook());\n\t\tbooks.add(createHorrorBook());\n\t\tbooks.add(createHorrorBook());\n\t\tbooks.add(createRomanceBook());\n\t\tbooks.add(createRomanceBook());\n\t\tbooks.add(createRomanceBook());\n\t\tbooks.add(createRomanceBook());\n\t\tbooks.add(createRomanceBook());\n\t\tbooks.add(createThrillerBook());\n\t\tbooks.add(createThrillerBook());\n\t\tbooks.add(createThrillerBook());\n\t\tbooks.add(createThrillerBook());\n\t\tbooks.add(createThrillerBook());\n\n\t\treturn books;\n\t}\n\n\tBook createFantasyBook() {\n\t\treturn createBook(\"\", Genre.fantasy);\n\t}\n\n\tBook createHorrorBook() {\n\t\treturn createBook(\"\", Genre.horror);\n\t}\n\n\tBook createRomanceBook() {\n\t\treturn createBook(\"\", Genre.romance);\n\t}\n\n\tBook createThrillerBook() {\n\t\treturn createBook(\"\", Genre.thriller);\n\t}\n\n\tBook createBook(String title, Genre genre) {\n\t\tBook book = new Book();\n\t\tbook.setBookId(nextBookId++);\n\t\tif (title == \"\") {\n\t\t\ttitle = \"# \" + Long.toString(book.getBookId());\n\t\t}\n\t\tbook.setTitle(title);\n\t\tbook.setGenre(genre);\n\n\t\treturn book;\n\t}\n}\n<\/pre>\ncreateBook<\/code> by passing the correct enum<\/code> type. Book ids start from 1001 and are set incrementally.ProducerChannelConfig<\/code> class that configures all the beans required for the producer application.<\/p>\npackage org.javacodegeeks.springintegration.kafka.config;\n\nimport java.util.HashMap;\nimport java.util.Map;\n\nimport org.apache.kafka.clients.producer.ProducerConfig;\nimport org.apache.kafka.common.serialization.StringSerializer;\nimport org.springframework.beans.factory.annotation.Value;\nimport org.springframework.context.annotation.Bean;\nimport org.springframework.context.annotation.Configuration;\nimport org.springframework.expression.common.LiteralExpression;\nimport org.springframework.integration.annotation.ServiceActivator;\nimport org.springframework.integration.channel.DirectChannel;\nimport org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;\nimport org.springframework.kafka.core.DefaultKafkaProducerFactory;\nimport org.springframework.kafka.core.KafkaTemplate;\nimport org.springframework.kafka.core.ProducerFactory;\nimport org.springframework.messaging.MessageHandler;\n\n@Configuration\npublic class ProducerChannelConfig {\n\n\t@Value(\"${spring.kafka.bootstrap-servers}\")\n\tprivate String bootstrapServers;\n\n\t@Bean\n\tpublic DirectChannel producerChannel() {\n\t\treturn new DirectChannel();\n\t}\n\n\t@Bean\n\t@ServiceActivator(inputChannel = \"producerChannel\")\n\tpublic MessageHandler kafkaMessageHandler() {\n\t\tKafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(kafkaTemplate());\n\t\thandler.setMessageKeyExpression(new LiteralExpression(\"kafka-integration\"));\n\n\t\treturn handler;\n\t}\n\n\t@Bean\n\tpublic KafkaTemplate kafkaTemplate() {\n\t\treturn new KafkaTemplate(producerFactory());\n\t}\n\n\t@Bean\n\tpublic ProducerFactory producerFactory() {\n\t\treturn new DefaultKafkaProducerFactory(producerConfigs());\n\t}\n\n\t@Bean\n\tpublic Map producerConfigs() {\n\t\tMap properties = new HashMap();\n\t\tproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);\n\t\tproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);\n\t\tproperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);\n\t\t\/\/ introduce a delay on the send to allow more messages to accumulate\n\t\tproperties.put(ProducerConfig.LINGER_MS_CONFIG, 1);\n\n\t\treturn properties;\n\t}\n}\n<\/pre>\nLibrary<\/code> class that is the main class of the application and the publisher endpoint of the system.<\/p>\npackage org.javacodegeeks.springintegration.kafka;\n\nimport java.util.Collections;\nimport java.util.List;\nimport java.util.Map;\n\nimport org.javacodegeeks.springintegration.kafka.incoming.BookPublisher;\nimport org.javacodegeeks.springintegration.kafka.model.Book;\nimport org.springframework.beans.factory.annotation.Autowired;\nimport org.springframework.boot.autoconfigure.SpringBootApplication;\nimport org.springframework.boot.builder.SpringApplicationBuilder;\nimport org.springframework.context.ConfigurableApplicationContext;\nimport org.springframework.kafka.support.KafkaHeaders;\nimport org.springframework.messaging.MessageChannel;\nimport org.springframework.messaging.support.GenericMessage;\n\n@SpringBootApplication\npublic class Library {\n\n\t@Autowired\n\tprivate BookPublisher bookPublisher;\n\n\tpublic static void main(String[] args) {\n\t\tConfigurableApplicationContext context = new SpringApplicationBuilder(Library.class).web(false).run(args);\n\t\tcontext.getBean(Library.class).run(context);\n\t\tcontext.close();\n\t}\n\n\tprivate void run(ConfigurableApplicationContext context) {\n\n\t\tSystem.out.println(\"Inside ProducerApplication run method...\");\n\n\t\tMessageChannel producerChannel = context.getBean(\"producerChannel\", MessageChannel.class);\n\n\t\tList books = bookPublisher.getBooks();\n\n\t\tfor (Book book : books) {\n\t\t\tMap headers = Collections.singletonMap(KafkaHeaders.TOPIC, book.getGenre().toString());\n\t\t\tproducerChannel.send(new GenericMessage(book.toString(), headers));\n\t\t}\n\n\t\tSystem.out.println(\"Finished ProducerApplication run method...\");\n\t};\n}\n<\/pre>\nMessageChannel<\/code> bean is obtained. It then takes a list of 20 books from BookPublisher<\/code> and sends those to producerChannel<\/code> which is wired to the Kafka broker. The topic of each message is the book genre.<\/p>\napplication.properties<\/code> file that specifies values to the environment variables.<\/p>\nspring.kafka.bootstrap-servers=localhost:9092\n<\/pre>\n
reader<\/b>: This is also a Maven-based project and all dependencies are configured in pom.xml.<\/p>\n<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<project xmlns=\"http:\/\/maven.apache.org\/POM\/4.0.0\" xmlns:xsi=\"http:\/\/www.w3.org\/2001\/XMLSchema-instance\"\nxsi: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>org.javacodegeeks.springintegration.kafka<\/groupId>\n<artifactId>consumer<\/artifactId>\n<version>0.0.1-SNAPSHOT<\/version>\n<packaging>jar<\/packaging>\n\n<name>consumer<\/name>\n<description>Kafka consumer with Spring Boot<\/description>\n\n<parent>\n <groupId>org.springframework.boot<\/groupId>\n <artifactId>spring-boot-starter-parent<\/artifactId>\n <version>1.5.9.RELEASE<\/version>\n <relativePath \/> <!-- lookup parent from repository -->\n<\/parent>\n\n<properties>\n <project.build.sourceEncoding>UTF-8<\/project.build.sourceEncoding>\n <project.reporting.outputEncoding>UTF-8<\/project.reporting.outputEncoding>\n <java.version>1.8<\/java.version>\n<\/properties>\n\n<dependencies>\n <dependency>\n <groupId>org.apache.kafka<\/groupId>\n <artifactId>kafka-clients<\/artifactId>\n <version>1.0.0<\/version>\n <\/dependency>\n <dependency>\n <groupId>org.springframework.boot<\/groupId>\n <artifactId>spring-boot-starter-integration<\/artifactId>\n <\/dependency>\n <dependency>\n <groupId>org.springframework.integration<\/groupId>\n <artifactId>spring-integration-kafka<\/artifactId>\n <version>2.3.0.RELEASE<\/version>\n <\/dependency>\n <dependency>\n <groupId>org.springframework.kafka<\/groupId>\n <artifactId>spring-kafka<\/artifactId>\n <version>1.3.2.RELEASE<\/version>\n <\/dependency>\n <dependency>\n <groupId>org.springframework.boot<\/groupId>\n <artifactId>spring-boot-starter-test<\/artifactId>\n <scope>test<\/scope>\n <\/dependency>\n<\/dependencies>\n\n<build>\n <plugins>\n <plugin>\n <groupId>org.springframework.boot<\/groupId>\n <artifactId>spring-boot-maven-plugin<\/artifactId>\n <\/plugin>\n <\/plugins>\n<\/build>\n<\/project>\n<\/pre>\n
ConsumerChannelConfig<\/code> class that configures all the beans required for the consumer application.[ulp id=’7POIYxRf1FUtPpmL’]<\/p>\npackage org.javacodegeeks.springintegration.kafka.config;\n\nimport java.util.HashMap;\nimport java.util.Map;\n\nimport org.apache.kafka.clients.consumer.ConsumerConfig;\nimport org.apache.kafka.common.serialization.StringDeserializer;\nimport org.springframework.beans.factory.annotation.Value;\nimport org.springframework.context.annotation.Bean;\nimport org.springframework.context.annotation.Configuration;\nimport org.springframework.integration.channel.QueueChannel;\nimport org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;\nimport org.springframework.kafka.core.ConsumerFactory;\nimport org.springframework.kafka.core.DefaultKafkaConsumerFactory;\nimport org.springframework.kafka.listener.ConcurrentMessageListenerContainer;\nimport org.springframework.kafka.listener.config.ContainerProperties;\nimport org.springframework.messaging.PollableChannel;\n\n@Configuration\npublic class ConsumerChannelConfig {\n\n\t@Value(\"${spring.kafka.bootstrap-servers}\")\n\tprivate String bootstrapServers;\n\n\t@Value(\"${spring.kafka.topic}\")\n\tprivate String springIntegrationKafkaTopic;\n\n\t@Bean\n\tpublic PollableChannel consumerChannel() {\n\t\treturn new QueueChannel();\n\t}\n\n\t@Bean\n\tpublic KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter() {\n\t\tKafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = new KafkaMessageDrivenChannelAdapter(\n\t\t\t\tkafkaListenerContainer());\n\t\tkafkaMessageDrivenChannelAdapter.setOutputChannel(consumerChannel());\n\n\t\treturn kafkaMessageDrivenChannelAdapter;\n\t}\n\n\t@SuppressWarnings(\"unchecked\")\n\t@Bean\n\tpublic ConcurrentMessageListenerContainer kafkaListenerContainer() {\n\t\tContainerProperties containerProps = new ContainerProperties(springIntegrationKafkaTopic);\n\n\t\treturn (ConcurrentMessageListenerContainer) new ConcurrentMessageListenerContainer(\n\t\t\t\tconsumerFactory(), containerProps);\n\t}\n\n\t@Bean\n\tpublic ConsumerFactory consumerFactory() {\n\t\treturn new DefaultKafkaConsumerFactory(consumerConfigs());\n\t}\n\n\t@Bean\n\tpublic Map consumerConfigs() {\n\t\tMap properties = new HashMap();\n\t\tproperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);\n\t\tproperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);\n\t\tproperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);\n\t\tproperties.put(ConsumerConfig.GROUP_ID_CONFIG, \"dummy\");\n\t\treturn properties;\n\t}\n}\n<\/pre>\nSubscribedReader<\/code> class that is the main class of the application and the consumer endpoint of the system.<\/p>\npackage org.javacodegeeks.springintegration.kafka;\n\nimport java.util.ArrayList;\nimport java.util.Arrays;\nimport java.util.List;\nimport java.util.Map;\n\nimport org.springframework.beans.factory.annotation.Autowired;\nimport org.springframework.boot.autoconfigure.SpringBootApplication;\nimport org.springframework.boot.autoconfigure.kafka.KafkaProperties;\nimport org.springframework.boot.builder.SpringApplicationBuilder;\nimport org.springframework.context.ConfigurableApplicationContext;\nimport org.springframework.integration.dsl.IntegrationFlow;\nimport org.springframework.integration.dsl.IntegrationFlows;\nimport org.springframework.integration.dsl.context.IntegrationFlowContext;\nimport org.springframework.integration.dsl.kafka.Kafka;\nimport org.springframework.kafka.core.DefaultKafkaConsumerFactory;\nimport org.springframework.messaging.Message;\nimport org.springframework.messaging.PollableChannel;\n\n@SpringBootApplication\npublic class SubscribedReader {\n\n\t@Autowired\n\tPollableChannel consumerChannel;\n\n\tpublic static void main(String[] args) {\n\n\t\tConfigurableApplicationContext context = new SpringApplicationBuilder(SubscribedReader.class).run(args);\n\n\t\tList valid_topics = Arrays.asList(\"fantasy\", \"horror\", \"romance\", \"thriller\");\n\n\t\tList topics = new ArrayList();\n\t\tif (args.length > 0) {\n\t\t\tfor (String arg : args) {\n\t\t\t\tif (valid_topics.contains(arg))\n\t\t\t\t\ttopics.add(arg);\n\t\t\t}\n\t\t}\n\n\t\tcontext.getBean(SubscribedReader.class).run(context, topics);\n\t\tcontext.close();\n\t}\n\n\tprivate void run(ConfigurableApplicationContext context, List topics) {\n\n\t\tSystem.out.println(\"Inside ConsumerApplication run method...\");\n\t\tPollableChannel consumerChannel = context.getBean(\"consumerChannel\", PollableChannel.class);\n\n\t\tfor (String topic : topics)\n\t\t\taddAnotherListenerForTopics(topic);\n\n\t\tMessage received = consumerChannel.receive();\n\t\twhile (received != null) {\n\t\t\treceived = consumerChannel.receive();\n\t\t\tSystem.out.println(\"Received \" + received.getPayload());\n\t\t}\n\t}\n\n\t@Autowired\n\tprivate IntegrationFlowContext flowContext;\n\n\t@Autowired\n\tprivate KafkaProperties kafkaProperties;\n\n\tpublic void addAnotherListenerForTopics(String... topics) {\n\t\tMap consumerProperties = kafkaProperties.buildConsumerProperties();\n\t\tIntegrationFlow flow = IntegrationFlows\n\t\t\t\t.from(Kafka.messageDrivenChannelAdapter(\n\t\t\t\t\t\tnew DefaultKafkaConsumerFactory(consumerProperties), topics))\n\t\t\t\t.channel(\"consumerChannel\").get();\n\t\tthis.flowContext.registration(flow).register();\n\t}\n}\n\n<\/pre>\nArrayList<\/code> that is passed as an argument to the run<\/code> method.<\/p>\nrun<\/code> method, a PollableChannel<\/code> bean, configured in ConsumerChannelConfig<\/code> is obtained from the application context. All the subscribed topics are added as listeners via MessageDrivenChannelAdapter<\/code> object by calling the method addAnotherListenerForTopics<\/code> for each topic. Then we call receive method of the PollableChannel<\/code> object inside a while loop to get the messages from the Kafka broker.<\/p>\napplication.properties<\/code> file that specifies values to the environment variables.<\/p>\nspring.kafka.bootstrap-servers=localhost:9092\nspring.kafka.topic=dummy\n<\/pre>\n
5. How to Run<\/h2>\n
Terminal 1: Start ZooKeeper. In your Kafka installation folder, run the following command:<\/p>\nbin\/zookeeper-server-start.sh config\/zookeeper.properties<\/pre>\n
bin\/kafka-server-start.sh config\/server.properties<\/pre>\n
mvn spring-boot:run -Dspring.kafka.consumer.group-id=\"group-one\" -Drun.arguments=\"fantasy,horror\"<\/pre>\n
mvn spring-boot:run -Dspring.kafka.consumer.group-id=\"group-two\" -Drun.arguments=\"horror,romance,thriller\"<\/pre>\n
mvn spring-boot:run<\/pre>\n
<\/a><\/a>6. Summary<\/h2>\n
7. Useful Links<\/h2>\n
\n
8. More articles<\/h2>\n