package com.grupohuman.settings.config
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer
import org.springframework.beans.factory.annotation.Value
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.kafka.annotation.EnableKafka
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
import org.springframework.kafka.config.KafkaListenerContainerFactory
import org.springframework.kafka.core.ConsumerFactory
import org.springframework.kafka.core.DefaultKafkaConsumerFactory
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer
import org.springframework.kafka.support.serializer.JsonDeserializer
import org.springframework.kafka.listener.KafkaListenerErrorHandler
@Configuration
@EnableKafka
class KafkaConsumerConfiguration {
  @Value('${spring.kafka.bootstrapServer}')
  private String bootstrapServers
  @Bean
  Map<String, Object> consumerConfigs() {
    Map<String, Object> props = [:]
    props.put ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers
    props.put ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer
    props.put ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer
    props.put ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString()
    props
  @Bean
  ConsumerFactory<String, String> consumerFactory() {
    new DefaultKafkaConsumerFactory<>(consumerConfigs(), null, new JsonDeserializer(Object))
  @Bean
  KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>()
    factory.setConsumerFactory(consumerFactory())
    factory
Producer Configuration:
package com.grupohuman.settings.config
import org.apache.kafka.clients.producer.ProducerConfig
import org.apache.kafka.common.serialization.StringSerializer
import org.springframework.beans.factory.annotation.Value
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.kafka.core.DefaultKafkaProducerFactory
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.kafka.core.ProducerFactory
import org.springframework.kafka.support.serializer.JsonSerializer
@Configuration
class KafkaProducerConfiguration {
    @Value('${spring.kafka.bootstrapServer}')
    private String bootstrapServer
    @Bean
    Map<String, Object> producerConfigs() {
        Map<String, Object> props = [:]
        props.put ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer
        props.put ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer
        props.put ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer
        props
    @Bean
    ProducerFactory<String, String> producerFactory() {
        new DefaultKafkaProducerFactory<>(producerConfigs())
    @Bean
    KafkaTemplate<String, String> kafkaTemplate() {
        new KafkaTemplate<>(producerFactory())
Consumer class:
package com.grupohuman.settings.kafka
import com.grupohuman.settings.entity.BranchOffice
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.springframework.kafka.annotation.KafkaListener
import org.springframework.stereotype.Component
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.messaging.handler.annotation.SendTo
import com.grupohuman.settings.service.BranchOfficeService
@Component
class BranchOfficeConsumer {
  @Autowired
  private BranchOfficeService branchOfficeService
  @KafkaListener(topics = 'branchOffice.findAll.start')
  @SendTo('branchOffice.findAll.end')
  List<BranchOffice> findAll() {
    branchOfficeService.findAll()
  @KafkaListener(topics = 'branchOffice.findById.start')
  @SendTo('branchOffice.findById.end')
  BranchOffice findById(BranchOffice entity) {
    branchOfficeService.findById entity.id
Main Application:
package com.grupohuman.settings
import org.springframework.boot.SpringApplication
import org.springframework.boot.autoconfigure.SpringBootApplication
@SpringBootApplication
class SettingsApplication {
	static void main(String[] args) {
		SpringApplication.run SettingsApplication, args
build.gradle
buildscript {
  ext {
    springBootVersion = '1.5.10.RELEASE'
  repositories {
    mavenCentral()
  dependencies {
    classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
apply plugin: 'groovy'
apply plugin: 'eclipse'
apply plugin: 'org.springframework.boot'
group = 'com.grupohuman.settings'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = 1.8
repositories {
  mavenCentral()
ext { springCloudVersion = 'Edgware.RELEASE' }
springBoot {
  executable = true
configurations {
  compile.exclude module: "spring-boot-starter-tomcat"
dependencies {
  compile 'org.springframework.boot:spring-boot-starter-data-jpa'
  compile 'org.springframework.cloud:spring-cloud-stream-binder-kafka'
  compile 'org.flywaydb:flyway-core'
  compile 'org.codehaus.groovy:groovy'
  compile 'org.postgresql:postgresql:9.4-1202-jdbc41'
  compile 'org.springframework.kafka:spring-kafka:2.1.2.RELEASE'
  runtime 'org.springframework.boot:spring-boot-devtools'
  testCompile 'org.springframework.boot:spring-boot-starter-test'
  testCompile 'junit:junit'
  testCompile 'com.h2database:h2'
  testCompile 'org.spockframework:spock-core:1.1-groovy-2.4'
  testCompile 'org.spockframework:spock-spring:1.1-groovy-2.4'
dependencyManagement {
  imports {
    mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}"
ERROR:




    

:compileJava NO-SOURCE
:compileGroovy
:processResources UP-TO-DATE
:classes
:findMainClass
:bootRun
13:08:52.219 [main] DEBUG org.springframework.boot.devtools.settings.DevToolsSettings - Included patterns for restart : []
13:08:52.223 [main] DEBUG org.springframework.boot.devtools.settings.DevToolsSettings - Excluded patterns for restart : [/spring-boot-starter/target/classes/, /spring-boot-autoconfigure/target/classes/, /spring-boot-starter-[\w-]+/, /spring-boot/target/classes/, /spring-boot-actuator/target/classes/, /spring-boot-devtools/target/classes/]
13:08:52.223 [main] DEBUG org.springframework.boot.devtools.restart.ChangeableUrls - Matching URLs for reloading : [file:/home/jeduardo/Desarrollo/0317humankhorpreferencias/build/classes/groovy/main/, file:/home/jeduardo/Desarrollo/0317humankhorpreferencias/build/resources/main/]
  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::       (v1.5.10.RELEASE)
2018-02-10 13:08:52 - Starting SettingsApplication on jorge.ramon with PID 9586 (/home/jeduardo/Desarrollo/0317humankhorpreferencias/build/classes/groovy/main started by jeduardo in /home/jeduardo/Desarrollo/0317humankhorpreferencias)
2018-02-10 13:08:52 - No active profile set, falling back to default profiles: default
2018-02-10 13:08:52 - Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@41bae31c: startup date [Sat Feb 10 13:08:52 CST 2018]; root of context hierarchy
2018-02-10 13:08:54 - No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.
2018-02-10 13:08:54 - No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
2018-02-10 13:08:54 - No bean named 'taskScheduler' has been explicitly defined. Therefore, a default ThreadPoolTaskScheduler will be created.
2018-02-10 13:08:54 - Bean 'org.springframework.kafka.annotation.KafkaBootstrapConfiguration' of type [org.springframework.kafka.annotation.KafkaBootstrapConfiguration$$EnhancerBySpringCGLIB$$9f6fa8e8] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2018-02-10 13:08:55 - Bean 'org.springframework.transaction.annotation.ProxyTransactionManagementConfiguration' of type [org.springframework.transaction.annotation.ProxyTransactionManagementConfiguration$$EnhancerBySpringCGLIB$$cbbd9765] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2018-02-10 13:08:55 - Bean 'integrationGlobalProperties' of type [org.springframework.beans.factory.config.PropertiesFactoryBean] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2018-02-10 13:08:55 - Bean 'integrationGlobalProperties' of type [java.util.Properties] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2018-02-10 13:08:55 - Flyway 3.2.1 by Boxfuse
2018-02-10 13:08:56 - Database: jdbc:postgresql://localhost:5432/khor2?currentSchema=preferencias (PostgreSQL 10.1)
2018-02-10 13:08:56 - Validated 4 migrations (execution time 00:00.033s)
2018-02-10 13:08:56 - Current version of schema "preferencias": 4
2018-02-10 13:08:56 - Schema "preferencias" is up to date. No migration necessary.
2018-02-10 13:08:56 - Building JPA container EntityManagerFactory for persistence unit 'default'
2018-02-10 13:08:56 - HHH000204: Processing PersistenceUnitInfo [
	name: default
2018-02-10 13:08:56 - HHH000412: Hibernate Core {5.0.12.Final}
2018-02-10 13:08:56 - HHH000206: hibernate.properties not found
2018-02-10 13:08:56 - HHH000021: Bytecode provider name : javassist
2018-02-10 13:08:56 - HCANN000001: Hibernate Commons Annotations {5.0.1.Final}
2018-02-10 13:08:56 - HHH000400: Using dialect: org.hibernate.dialect.PostgreSQLDialect
2018-02-10 13:08:57 - HHH000424: Disabling contextual LOB creation as createClob() method threw error : java.lang.reflect.InvocationTargetException
2018-02-10 13:08:57 - HHH000270: Type registration [java.util.UUID] overrides previous : org.hibernate.type.UUIDBinaryType@30ef1685
2018-02-10 13:08:58 - Initialized JPA EntityManagerFactory for persistence unit 'default'
2018-02-10 13:08:59 - Initializing ExecutorService  'taskScheduler'
2018-02-10 13:08:59 - Unable to start LiveReload server
2018-02-10 13:08:59 - configured Kryo registration [40, java.io.File] with serializer org.springframework.integration.codec.kryo.FileSerializer
2018-02-10 13:09:00 - 
Error starting ApplicationContext. To display the auto-configuration report re-run your application with 'debug' enabled.
2018-02-10 13:09:00 - Application startup failed
java.lang.NoSuchMethodError: org.springframework.util.Assert.state(ZLjava/util/function/Supplier;)V
	at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.determineInferredType(MessagingMessageListenerAdapter.java:396)
	at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.<init>(MessagingMessageListenerAdapter.java:100)
	at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.<init>(RecordMessagingMessageListenerAdapter.java:61)
	at org.springframework.kafka.config.MethodKafkaListenerEndpoint.createMessageListenerInstance(MethodKafkaListenerEndpoint.java:172)
	at org.springframework.kafka.config.MethodKafkaListenerEndpoint.createMessageListener(MethodKafkaListenerEndpoint.java:132)
	at org.springframework.kafka.config.AbstractKafkaListenerEndpoint.setupMessageListener(AbstractKafkaListenerEndpoint.java:355)
	at org.springframework.kafka.config.AbstractKafkaListenerEndpoint.setupListenerContainer(AbstractKafkaListenerEndpoint.java:340)
	at org.springframework.kafka.config.AbstractKafkaListenerContainerFactory.createListenerContainer(AbstractKafkaListenerContainerFactory.java:227)
	at org.springframework.kafka.config.AbstractKafkaListenerContainerFactory.createListenerContainer(AbstractKafkaListenerContainerFactory.java:49)
	at org.springframework.kafka.config.KafkaListenerEndpointRegistry.createListenerContainer(KafkaListenerEndpointRegistry.java:183)
	at org.springframework.kafka.config.KafkaListenerEndpointRegistry.registerListenerContainer(KafkaListenerEndpointRegistry.java:155)
	at org.springframework.kafka.config.KafkaListenerEndpointRegistry.registerListenerContainer(KafkaListenerEndpointRegistry.java:129)
	at org.springframework.kafka.config.KafkaListenerEndpointRegistrar.registerAllEndpoints(KafkaListenerEndpointRegistrar.java:138)
	at org.springframework.kafka.config.KafkaListenerEndpointRegistrar.afterPropertiesSet(KafkaListenerEndpointRegistrar.java:132)
	at org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor.afterSingletonsInstantiated(KafkaListenerAnnotationBeanPostProcessor.java:243)
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:781)
	at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:867)
	at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:543)
	at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:693)
	at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:360)
	at org.springframework.boot.SpringApplication.run(SpringApplication.java:303)
	at org.springframework.boot.SpringApplication.run(SpringApplication.java:1118)
	at org.springframework.boot.SpringApplication.run(SpringApplication.java:1107)
	at org.springframework.boot.SpringApplication$run.call(Unknown Source)
	at org.codehaus.groovy.runtime.callsite.CallSiteArray.defaultCall(CallSiteArray.java:48)
	at org.codehaus.groovy.runtime.callsite.AbstractCallSite.call(AbstractCallSite.java:113)
	at org.codehaus.groovy.runtime.callsite.AbstractCallSite.call(AbstractCallSite.java:133)
	at com.grupohuman.settings.SettingsApplication.main(SettingsApplication.groovy:10)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.springframework.boot.devtools.restart.RestartLauncher.run(RestartLauncher.java:49)
2018-02-10 13:09:00 - Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@41bae31c: startup date [Sat Feb 10 13:08:52 CST 2018]; root of context hierarchy
2018-02-10 13:09:00 - Shutting down ExecutorService 'taskScheduler'
2018-02-10 13:09:00 - Unregistering JMX-exposed beans on shutdown
2018-02-10 13:09:00 - Unregistering JMX-exposed beans on shutdown
2018-02-10 13:09:00 - Closing JPA EntityManagerFactory for persistence unit 'default'
BUILD SUCCESSFUL in 12s
4 actionable tasks: 3 executed, 1 up-to-date