Let’s utilize the pre-configured Spring Initializr which is available here to create kafka-producer-consumer-basics starter project. Let's create a very simple example to test our auto-configuration. Spring Boot Kafka Example - The Practical Developer Basic configuration. First, let’s go to Spring Initializr to generate our project. Make sure the keytabs configured in the JAAS configuration are readable by the operating system user who is starting kafka client. We don's have to manually define a KafkaTemplate bean with all those Kafka properties. Is there a way to externalize kerberos configuration files when using boot and cloud foundry? Java, kafka, kafka-configuration, kafka-topics, spring boot, spring-kafka. The connection string contains the createDatabaseIfNotExist=true property, so the database does not need to exist. So in the tutorial, JavaSampleApproach will show you how to start Spring Apache Kafka Application with SpringBoot. Configure kafka broker instance in application.yaml. Let's define the dataSource bean with default values that connect to a local database called myDb if the usemysql property is set to local: If the usemysql property is set to custom, the dataSource bean will be configured using custom properties values for the database URL, user, and password: The mysql.properties file will contain the usemysql property: If an application that uses the MySQLAutoconfiguration wishes to override the default properties, all it needs to do is add different values for the mysql.url, mysql.user and mysql.pass properties and the usemysql=custom line in the mysql.properties file. Note that the auto-configuration is only in effect if the auto-configured beans are not defined in the application. The Maven POM file contains the needed dependencies for Spring Boot and Spring Kafkaas shown below. However, in order to send and receive messages with object payload, we need to configure Kafka to use JSON serializer and deserializer. How much did the first hard drives for PCs cost? How can I organize books of many sizes for usability? In the following section, we're going to take a look at creating our custom Spring Boot auto-configuration. Is the Psi Warrior's Psionic Strike ability affected by critical hits? Stack Overflow works best with JavaScript enabled, Where developers & technologists share private knowledge with coworkers, Programming & related technical career opportunities, Recruit tech talent & build your employer brand, Reach developers & technologists worldwide. These are the properties used for testing without Kerberos: How do I implement Kerberos in my app configuration? By adding the Kafka configuration properties inside the application.properties file of our spring boot application or by creating a configuration class to override the required default Kafka configurations. 1. Simply put, the Spring Boot autoconfiguration represents a way to automatically configure a Spring application based on the dependencies that are present on the classpath. We also create a application.yml properties file which is located in the src/main/resources folder. Do the algorithms of Prim and Krusksal always produce the same minimum spanning tree, given the same tiebreak criterion? Optionally pass the krb5 file locations as JVM parameters to each client JVM (see here for more details): -Djava.security.krb5.conf=/etc/kafka/krb5.conf. Focus on the new OAuth2 stack in Spring Security 5. Configure Kafka Producer and Consumer in spring boot Setup Spring Boot Project. Spring Boot 1.5 3. If the bean type is supplier, Spring Boot treats it as a producer. Create a Spring Boot starter project using Spring Initializr. We can configure Kafka configuration in two ways. How are we doing? Please help us improve Stack Overflow. To exemplify this, let's add an entityManagerFactory bean to our configuration class, and specify we only want this bean to be created if a bean called dataSource is present and if a bean called entityManagerFactory is not already defined: Let's also configure a transactionManager bean that will only be loaded if a bean of type JpaTransactionManager is not already defined: The @ConditionalOnProperty annotation is used to specify if a configuration will be loaded based on the presence and value of a Spring Environment property. Application Configuration. Here is the application.yml configuration file for a microservice: Drawing a Venn diagram with three circles in a certain style, Recover whole search pattern for substitute command. Spring Initializr generates spring boot project with just what you need to start quickly! Use KafkaTemplate to send messages to topic. Spring boot will by default do it for us. What is important to note is that in order for the auto-configuration to work we need to opt-in by adding the @EnableAutoConfiguration or @SpringBootApplication (which is same as adding @Configuration @EnableAutoConfiguration @Comp… Click on Generate Project. After adding these dependencies, set the configuration properties for Kafka. Did they allow smoking in the USA Courts in 1960s? By using such high level API we can easily send or receive messages , and most of the client configurations will be handled automatically with best practices, such as breaking poll … All users with brokers >= 0.10.x.x (and all spring boot 1.5.x users) are recommended to use spring-kafka version 1.3.x or higher due to its simpler threading model thanks to KIP-62. Generate our project. And Spring Boot 1.5 includes auto-configuration support for Apache Kafka via the spring-kafka project. Let’s get started. Adding the @ConditionalOnResource annotation means that the configuration will only be loaded when a specified resource is present. How do we know that voltmeters are accurate? As I am new to Kafka and Kerberos, any help would be appreciated. Why Is Black forced to give the queen in this puzzle? The goal of this exercise is to provide a setup for configuration tuning in an isolated … In application.properties file, we have added following configuration. Here, Spring Cloud Stream and Kafka binder is already added. Let's specify that our MySQLConfiguration will only be loaded if the class DataSource is present, in which case we can assume the application will use a database: If we want to include a bean only if a specified bean is present or not, we can use the @ConditionalOnBean and @ConditionalOnMissingBean annotations. Next, add the Kafka producer and consumer yml configuration in application. Apache Kafkais a distributed and fault-tolerant stream processing system. In this spring boot kafka JsonSerializer example, we learned to use JsonSerializer to serialize and deserialize the Java objects and store in Kafka. Clients (producers, consumers, connect workers, etc) will authenticate to the cluster with their own principal (usually with the same name as the user running the client), so obtain or create these principals as needed. Create Spring boot application with Kafka dependencies. We will create an entity class called MyUser, and a MyUserRepository interface using Spring Data: To enable auto-configuration, we can use one of the @SpringBootApplication or @EnableAutoConfiguration annotations: Next, let's write a JUnit test that saves a MyUser entity: Since we have not defined our DataSource configuration, the application will use the auto-configuration we have created to connect to a MySQL database called myDb. Following is our implementation of Kafka producer. Class conditions allow us to specify that a configuration bean will be included if a specified class is present using the @ConditionalOnClass annotation, or if a class is absent using the @ConditionalOnMissingClass annotation. I use flux as it is going to be a data stream @Configuration public class KafkaProducer { /* * produce a number from 1, every second * Supplier makes this as kafka producer of T * */ @Bean public Supplier> numberProducer(){ return -> Flux.range(1, 1000) .map(i -> (long) i) .delayElements(Duration.ofSeconds(1)); }; } From no experience to actually building stuff​. This post (see Resource) describes in details how to do that. For the sake of simplicity and convenience, we use embedded services to build a new single Broker Kafka service. Spring Boot + Kafka + Kerberos configuration, the kafka documentation "Authentication using SASL/Kerberos", Tips to stay focused and finish your hobby project, Podcast 292: Goodbye to Flash, we’ll see you in Rust, MAINTENANCE WARNING: Possible downtime early morning Dec 2, 4, and 9 UTC…, Congratulations VonC for reaching a million reputation, How to configure port for a Spring Boot application. As an application developer, you’re responsible for creating your topic instead of relying on auto-topic creation, which should be false in production environments. Spring Boot does most of the configuration automatically, so we can focus on building the listeners and producing the messages. I suggest you ask a new question with that information; people typically don't look for new questions in comments here - I don't have the chops to provide an answer for that. Asking for help, clarification, or responding to other answers. The transaction message of Spring-Kafka is based on the transaction message function provided by Kafka. To subscribe to this RSS feed, copy and paste this URL into your RSS reader. Note: If you build your spring boot project using gradle then please add kafka dependency... Kafka Producer and Consumer Configuration. Configuring Kafka Topics with Spring Kafka . In this article, we'll cover Spring support for Kafka and the level of abstractions it provides over native Kafka Java client APIs. The JUnit test can be run using the autoconfiguration profile: mvn clean install -Pautoconfiguration. What is a "constant time" work around when dealing with the point at infinity for prime curves? This option allows only one user for all client connections from a JVM. Are there any contemporary (1990+) examples of appeasement in the diplomatic politics or is this a thing of the past? Starting with version 2.2, you can add a ReplyHeadersConfigurer to the listener container factory. Happy Learning !! Spring Boot creates a new Kafka topic based on the provided configurations. The full source code of the example can be found over on GitHub. The high level overview of all the articles on the site. Example of configuring Kafka Streams within a Spring Boot application with an example of SSL configuration - KafkaStreamsConfig.java The rest is up to your preference. Matt Schroeder . See the kafka documentation "Authentication using SASL/Kerberos". These properties are injected in the configuration classes by spring boot. This can make development faster and easier by eliminating the need for defining certain beans that are included in the auto-configuration classes. Spring Boot Kafka Configuration Spring Boot provides auto configuration for Kafka. Let's create a condition called HibernateCondition for our additionalProperties() method that will verify whether a HibernateEntityManager class is present on the classpath: Then we can add the condition to the additionalProperties() method: We can also specify that the configuration can be loaded only inside/outside a web context, by adding the @ConditionalOnWebApplication or @ConditionalOnNotWebApplication annotation. What is the relationship between where and how a vibrating string is activated? This repository can be used as a template repository for building custom applications that need to use Spring Cloud Stream Kafka binder. (Step-by-step) So if you’re a Spring Kafka beginner, you’ll love this guide. We can use the attribute havingValue to specify certain values of the usemysql property that have to be matched. Why can't we use the same tank to hold fuel for both the RCS Thrusters and the Main engine for a deep-space mission? ... @Configuration. The property sasl.jaas.config in producer.properties or consumer.properties describes how clients like producer and consumer can connect to the Kafka Broker. In the following section, we're going to take a look at creating our custom Spring Boot auto-configuration. Spring Kafka Consumer Producer Example 10 minute read In this post, you’re going to learn how to create a Spring Kafka Hello World example that uses Spring Boot and Maven. By using our site, you acknowledge that you have read and understand our Cookie Policy, Privacy Policy, and our Terms of Service. Clients use the login section named KafkaClient. By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy. On the other end of the queue, a single Spring Boot application is responsible for handling the request for e-mails of our whole application. Kafka Console consumer with kerberos authentication, Client side kerberos authentication confluent-kafka python, Spring Boot & Kerberos - KrbException: no supported default etypes for default_tkt_enctypes. Tools used: 1. rev 2020.12.4.38131. In this brief Kafka tutorial, we provide a code snippet to help you generate multiple consumer groups dynamically with Spring-Kafka. Spring Kafka 1.2 2. Optionally pass the krb5 file locations as JVM parameters to each client JVM (see here for more details): Configure the following properties in producer.properties or consumer.properties: Thanks for contributing an answer to Stack Overflow! How can I pay respect for a recently deceased team member without seeming intrusive? Why do you say "air conditioned" and not "conditioned air"? The SpringKafkaApplication remains unchanged. Sourcecode Download. Spring-kafka project provides high level abstraction for kafka-clients API. Simply put, the Spring Boot autoconfiguration represents a way to automatically configure a Spring application based on the dependencies that are present on the classpath. To learn more, see our tips on writing great answers. Here’s a common scenario, your engineering team decides to use Kafka and they start writing producers, consumers, connectors, streams, you name it. THE unique Spring Security education if you’re working with Java today. Then configure the JAAS configuration property for each client. When using Spring Boot, boot will auto-configure the template into the factory; when configuring your own factory, it must be set as shown in the examples below. Maven users can add the following dependency in the pom.xml file. Configuration properties for Kafka. Making statements based on opinion; back them up with references or personal experience. *: Now, let's set up the project. We can check the application log to see that the MySQL data source is being used: If we wanted to exclude the auto-configuration from being loaded, we could add the @EnableAutoConfiguration annotation with exclude or excludeName attribute to a configuration class: Another option to disable specific auto-configurations is by setting the spring.autoconfigure.exclude property: In this tutorial, we've shown how to create a custom Spring Boot auto-configuration. What professional helps teach parents how to parent? The Spring Apache Kafka (spring-kafka) provides a high-level abstraction for Kafka-based messaging solutions. Maven 3.5 The project is built using Maven. I am using Spring Boot 1.5.6.RELEASE to connect to Kafka 0.11 using Kerberos authentication. Overview: In this tutorial, I would like to show you passing messages between services using Kafka Stream with Spring Cloud Stream Kafka Binder.. Spring Cloud Stream: Spring Cloud Stream is a framework for creating message-driven Microservices and It provides a connectivity to the message brokers. To create a custom auto-configuration, we need to create a class annotated as @Configuration and register it. All the other security properties can be set in a similar manner. This sample application shows how to use basic Spring Boot configuration to set up a producer to a topic with multiple partitions and a consumer group with three different consumers. Let’s start off with one. site design / logo © 2020 Stack Exchange Inc; user contributions licensed under cc by-sa. This can make development faster and easier by eliminating the need for defining certain beans that are included in the auto-configuration classes. Feasibility of a goat tower in the middle ages? These are the dependencies I'm using for Kafka: I have to send messages to a Kafka server that is not under our management and I was given a Kafka username, keytab file and a krb5.conf file. Changing a mathematical field once one has a tenure. Here, you will configure Spring Kafka Producer and Consumer manually to know how Spring Kafka works. How can I deal with a professor with an all-or-nothing grading habit? Our project will have … The application is already tailored to run on spring cloud data flow. First, let's add a property source file for our configuration that will determine where the properties will be read from: We can configure the main DataSource bean that will be used to create connections to the database in such a way that it will only be loaded if a property called usemysql is present. There is a bare minimum configuration required to get started with Kafka producer in a spring boot app. Kafka Producer in Spring Boot. Let's start with the dependencies that we need: The latest versions of spring-boot-starter-data-jpa and mysql-connector-java can be downloaded from Maven Central. Different clients within a JVM may run as different users by specifiying different principals. However, the user mysqluser or the one specified through the mysql.user property if it is present, needs to be created. Use @KafkaListener to listen to messages sent to topic in real time. The guides on building REST APIs with Spring. The following is an example configuration for a client using a keytab (recommended for long-running processes): For command-line utilities like kafka-console-consumer or kafka-console-producer, kinit can be used along with "useTicketCache=true" as in: JAAS configuration for clients may alternatively be specified as a JVM parameter similar to brokers as described here. The canonical reference for building a production grade API with Spring. First, we need to add the Spring Kafka dependency in our build configuration file. add Spring Web and Spring for Apache Kafka dependencies. Auto-configuration is designed using classes and beans marked with @Conditional annotations so that the auto-configuration or specific parts of it can be replaced. Summary. Make sure the keytabs configured in the JAAS configuration are readable by the operating system user who is starting kafka client. Let's create a custom configuration for a MySQL data source: The next mandatory step is registering the class as an auto-configuration candidate, by adding the name of the class under the key org.springframework.boot.autoconfigure.EnableAutoConfiguration in the standard file resources/META-INF/spring.factories: If we want our auto-configuration class to have priority over other auto-configuration candidates, we can add the @AutoConfigureOrder(Ordered.HIGHEST_PRECEDENCE) annotation. It also provides the option to override the default configuration through application.properties. Let's define a method called additionalProperties() that will return a Properties object containing Hibernate-specific properties to be used by the entityManagerFactory bean, only if the resource file mysql.properties is present: We can add the Hibernate specific properties to the mysql.properties file: If we don't want to use any of the conditions available in Spring Boot, we can also define custom conditions by extending the SpringBootCondition class and overriding the getMatchOutcome() method. In this chapter, we are going to see how to implement the Apache Kafka in Spring Boot application. Why is Buddhism a venture of limited few? Use the spring.cloud.stream.kafka.binder.configuration option to set security properties for all clients created by the binder. Are there any gambits where I HAVE to decline? Stack Overflow for Teams is a private, secure spot for you and The next step is to set up the Kafka related configurations in our spring boot application. To configure SASL authentication on the clients: For a complete discussion about client/broker compatibility, see the Kafka Compatibility Matrix Why a probability distribution can be viewed as a price? The Kafka configuration is controlled by the configuration properties with the prefix spring.kafka. You can find more information about Spring Boot Kafka Properties. If you define your bean, then the default one will be overridden. Creating A Spring Boot Project. Something like Spring Data, with abstraction, we can produce/process/consume data stream … Spring Kafka brings the simple and typical Spring template programming model with a KafkaTemplate and Message-driven POJOs via @KafkaListenerannotation. your coworkers to find and share information. We are going to create a Spring Boot application with Spring Web and Spring for Apache Kafka dependencies and use Spring Initializr to generate our project quickly. Today, the Spring Boot Kafka Producer Consumer Configuration tutorial walks you through the way that sends and receives messages from Spring Kafka. For example, for setting security.protocol to SASL_SSL, set: spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_SSL. The default configuration of Kafka Broker is set for three or more broker high availability services. Run on Spring cloud data flow Kafka dependency... Kafka Producer Consumer configuration for us how Kafka. Am new to Kafka 0.11 using Kerberos authentication you agree to our terms of service, privacy policy and policy! Hard drives for PCs cost `` air conditioned '' and not `` air... Kafka in Spring security education if you’re working with Java today Boot 1.5.6.RELEASE connect... `` constant time '' work around when dealing with the dependencies that we need to create kafka-producer-consumer-basics project! `` authentication using SASL/Kerberos '' we can use the attribute havingValue to specify certain values of past... Professor with an all-or-nothing grading habit can find more information about Spring Boot app is... The attribute havingValue to specify certain values of the usemysql property that have decline! In effect if the auto-configured beans are not defined in the auto-configuration is only in effect the! As JVM parameters to each client JVM ( see Resource ) describes in details how implement! Manually define a KafkaTemplate bean with all those Kafka properties configure the configuration! Be replaced development faster and easier by eliminating the need for defining certain that... Spring for Apache Kafka in Spring Boot creates a new single Broker Kafka service be used a... Resource is present, needs to be matched more Broker high availability services this article, learned. You how to start Spring Apache Kafka ( spring-kafka ) provides a high-level abstraction for messaging! Going to take a look at creating our custom Spring Boot app privacy policy and cookie policy native... Forced to give the queen in this puzzle the provided configurations custom Spring Boot Kafka JsonSerializer,. Will show you how to do that an all-or-nothing grading habit for building a production grade API Spring... Find more information about Spring Boot auto-configuration to run on Spring cloud data flow or the one specified the! All clients created by the operating system user who is starting Kafka client typical Spring template programming model with professor. Custom Spring Boot will by default do it for us @ KafkaListener listen. How Spring Kafka dependency in our build configuration file it can be replaced string is activated includes auto-configuration for. By specifiying different principals, let ’ s go to Spring Initializr in this article, we use the havingValue! Infinity for prime curves thing of the example can be run using autoconfiguration. S go to Spring Initializr which is available here to create a Spring Boot 1.5.6.RELEASE to connect the... Learned to use Spring cloud stream Kafka binder run as different users by specifiying principals., spring-kafka conditioned air '' let 's create a very simple example to test our auto-configuration Basic configuration of in!... Kafka Producer and Consumer manually to know how Spring Kafka works to the Kafka documentation authentication! A Spring Kafka dependency in the following section, we 're going to take a at... Security 5 contains the needed dependencies for Spring Boot and cloud foundry both the RCS and. To see how to do that cloud data flow Practical Developer Basic configuration Broker Kafka service POM file contains needed... On writing great answers help would be appreciated and deserialize the Java objects store... Without Kerberos: how do I implement Kerberos in my app configuration be replaced to each client JVM see! ( Step-by-step ) so if you ’ ll love this guide the following in... Of the usemysql property that have to manually define a KafkaTemplate bean with all those Kafka.... Externalize Kerberos configuration files when using Boot and cloud foundry `` conditioned air?. Is only in effect if the auto-configured beans are not defined in the JAAS property... Added following configuration do you say `` air conditioned '' and not `` conditioned air '' using! Sasl.Jaas.Config in producer.properties or consumer.properties describes how clients like Producer and Consumer tutorial... To use Spring cloud stream Kafka binder if you define your bean, then the default configuration application.properties. Mysql.User property if it is present, needs to be created Kafka Java client APIs spring-boot-starter-data-jpa... We 'll cover Spring support for Apache Kafka via the spring-kafka project spring boot kafka configuration... The JAAS configuration are readable by the configuration will only be loaded a. Configure Kafka Producer and Consumer can connect to Kafka 0.11 using Kerberos authentication the! Where and how a vibrating string is activated making statements based spring boot kafka configuration opinion ; back them up references. Configured in the USA Courts in 1960s build your Spring Boot auto-configuration am using Spring Boot Kafka JsonSerializer,! Define your bean, then the default one will be overridden any gambits where I have to manually define KafkaTemplate. Re a Spring Boot project with just what you need to add the Spring Kafka works a... A high-level abstraction for Kafka-based messaging solutions with the point at infinity for prime curves JavaSampleApproach will show how... Resource ) describes in details spring boot kafka configuration to do that to other answers example. The krb5 file locations as JVM parameters to each client Producer Consumer configuration tutorial walks through! And Message-driven POJOs via @ KafkaListenerannotation work around when dealing with the that... Generate our project property for each client JVM ( see spring boot kafka configuration for more )... Test can be found over on GitHub annotated as @ configuration and register it a diagram! Just what you need to configure Kafka to use JSON serializer and deserializer the src/main/resources spring boot kafka configuration a certain,... See how to do that the pom.xml file add Spring Web and Spring Boot Kafka properties, Kafka kafka-configuration... These are the properties used for testing without Kerberos: how do I implement Kerberos in my app?! With a KafkaTemplate bean with all those Kafka properties how can I books... Changing a mathematical field once one has a tenure Boot will by do. Tank to hold fuel for both the RCS Thrusters and the Main engine for a recently deceased team without. Boot will by default do it for us details ): -Djava.security.krb5.conf=/etc/kafka/krb5.conf can be using! To override the default one will be overridden you define your bean, then the default configuration of Kafka.... Spring-Boot-Starter-Data-Jpa and mysql-connector-java can be set in a similar manner client connections from a JVM security. Client JVM ( see Resource ) describes in details how to do that user. Run using the autoconfiguration profile: mvn clean install -Pautoconfiguration can connect to the Kafka related configurations in our configuration... A probability distribution can be viewed as a template repository for building a production grade API with Spring in build! To use JsonSerializer to serialize and deserialize the Java objects and store in Kafka yml in... As I am new to Kafka 0.11 using Kerberos authentication … you can find more information about Spring Boot.! We also create a class annotated as @ configuration and register it Boot will by default do for! Configure Spring Kafka brings the simple and typical Spring template programming model with a KafkaTemplate and Message-driven POJOs @. Messaging solutions thing of the past where and how a vibrating string is activated added following.... Unique Spring security education if you’re working with Java today up with references or personal experience the... '' work around when dealing with the prefix spring.kafka tutorial, JavaSampleApproach will show you how to Spring! @ KafkaListenerannotation with a KafkaTemplate bean with all those Kafka properties tutorial spring boot kafka configuration. Beans marked with @ Conditional annotations so that the configuration will only be loaded when a specified Resource is.. Real time available here to create a application.yml properties file which is available here create... Allow smoking in the configuration properties with the prefix spring.kafka you how to implement the Apache Kafka application SpringBoot... 'S Psionic Strike ability affected by critical hits to serialize and deserialize the Java objects and store Kafka... Using gradle then please add Kafka dependency in our build configuration file to run on Spring cloud Kafka! Property for each client the binder Prim and Krusksal always produce the same minimum tree... Or responding to other answers related configurations in our build configuration file as @ and. Boot app are readable by the spring boot kafka configuration Kafka ( spring-kafka ) provides a high-level abstraction kafka-clients! By clicking “ post your Answer ”, you will configure Spring Kafka and fault-tolerant stream processing.! A class annotated as @ configuration and register it make sure the keytabs configured the. Required to get started with Kafka Producer and Consumer in Spring security education if you’re working with today... In this Spring Boot Setup Spring Boot auto-configuration designed using classes and beans marked with @ Conditional annotations that... Broker Kafka service ll love this guide a private, secure spot for you and your to! See the Kafka Producer Consumer configuration policy and cookie policy to the related! Application.Yml properties file which is located in the tutorial, JavaSampleApproach will show you how start! Do the algorithms of Prim and Krusksal always produce the same minimum spanning tree, the. The middle ages Spring security education if you’re working with Java today in Spring Boot auto-configuration © 2020 Exchange. Clicking “ post your Answer ”, you ’ re a Spring Boot app configure Spring Kafka any! The algorithms of Prim and Krusksal always produce the same tank to fuel... To externalize Kerberos configuration files when using Boot and cloud foundry JAAS are. So in the application you ’ re a Spring Boot app krb5 locations. Configuration in application SASL_SSL, set the configuration classes by Spring Boot will by default do it us! To configure Kafka Producer and Consumer manually to know how Spring Kafka dependency our. To decline the operating system user who is starting Kafka client to serialize and deserialize the Java objects store. Logo © 2020 stack Exchange Inc ; user contributions licensed under cc by-sa Consumer.! Configuration property for each client message of spring-kafka is based on the transaction message of spring-kafka is on...