{"action":"opened","issue":{"url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/57","labels_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/57/labels{/name}","comments_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/57/comments","events_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/57/events","html_url":"https://github.com/spring-projects/spring-integration-kafka/issues/57","id":103145658,"number":57,"title":"Spring Kafka Integration: Consumer code is not getting invoked.","user":{"login":"suchitgupta","id":7201142,"avatar_url":"https://avatars.githubusercontent.com/u/7201142?v=3","gravatar_id":"","url":"https://api.github.com/users/suchitgupta","html_url":"https://github.com/suchitgupta","followers_url":"https://api.github.com/users/suchitgupta/followers","following_url":"https://api.github.com/users/suchitgupta/following{/other_user}","gists_url":"https://api.github.com/users/suchitgupta/gists{/gist_id}","starred_url":"https://api.github.com/users/suchitgupta/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/suchitgupta/subscriptions","organizations_url":"https://api.github.com/users/suchitgupta/orgs","repos_url":"https://api.github.com/users/suchitgupta/repos","events_url":"https://api.github.com/users/suchitgupta/events{/privacy}","received_events_url":"https://api.github.com/users/suchitgupta/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":0,"created_at":"2015-08-26T00:44:47Z","updated_at":"2015-08-26T00:44:47Z","closed_at":null,"body":"Hi,\r\n\r\nI am building up  using Spring MVC and Spring Kaka Integration. The code on a URL hit produces the messages and once a message is send I want to invoke my consumer code so that It can absorb the message. While first part that is sending the message to broker is working fine but the remaining part that is to consume the message automatically is not working.\r\n\r\n1. Spring MVC / Spring Integration version is 4.1.5.Release.\r\n2. spring-integration-kafka is 1.2.0.Release.\r\n\r\nPom.xml:\r\n\r\n<project xmlns=\"http://maven.apache.org/POM/4.0.0\" xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\"\r\n\txsi:schemaLocation=\"http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd\">\r\n\t<modelVersion>4.0.0</modelVersion>\r\n\t<groupId>com.happystudio</groupId>\r\n\t<artifactId>KafKaWithSpringMVC</artifactId>\r\n\t<packaging>war</packaging>\r\n\t<version>0.0.1-SNAPSHOT</version>\r\n\t<name>Spring MVC Tutorial</name>\r\n\t<url>http://maven.apache.org</url>\r\n\t<properties>\r\n\t\t<spring.version>4.1.5.RELEASE</spring.version>\r\n\t</properties>\r\n\t<dependencies>\r\n\t\t<dependency>\r\n\t\t\t<groupId>javax.servlet</groupId>\r\n\t\t\t<artifactId>javax.servlet-api</artifactId>\r\n\t\t\t<version>3.1.0</version>\r\n\t\t\t<scope>provided</scope>\r\n\t\t</dependency>\r\n\t\t<dependency>\r\n\t\t\t<groupId>org.springframework</groupId>\r\n\t\t\t<artifactId>spring-core</artifactId>\r\n\t\t\t<version>${spring.version}</version>\r\n\t\t</dependency>\r\n\t\t<dependency>\r\n\t\t\t<groupId>org.springframework</groupId>\r\n\t\t\t<artifactId>spring-web</artifactId>\r\n\t\t\t<version>${spring.version}</version>\r\n\t\t</dependency>\r\n\t\t<dependency>\r\n\t\t\t<groupId>org.springframework</groupId>\r\n\t\t\t<artifactId>spring-webmvc</artifactId>\r\n\t\t\t<version>${spring.version}</version>\r\n\t\t</dependency>\r\n\t\t<dependency>\r\n\t\t\t<groupId>junit</groupId>\r\n\t\t\t<artifactId>junit</artifactId>\r\n\t\t\t<version>4.11</version>\r\n\t\t\t<scope>test</scope>\r\n\t\t</dependency>\r\n\t\t<dependency>\r\n\t\t\t<groupId>org.springframework.integration</groupId>\r\n\t\t\t<artifactId>spring-integration-kafka</artifactId>\r\n\t\t\t<version>1.2.0.RELEASE</version>\r\n\t\t\t<exclusions>\r\n\t\t\t\t<exclusion>\r\n\t\t\t\t\t<artifactId>jms</artifactId>\r\n\t\t\t\t\t<groupId>javax.jms</groupId>\r\n\t\t\t\t</exclusion>\r\n\t\t\t\t<exclusion>\r\n\t\t\t\t\t<artifactId>jmxri</artifactId>\r\n\t\t\t\t\t<groupId>com.sun.jmx</groupId>\r\n\t\t\t\t</exclusion>\r\n\t\t\t\t<exclusion>\r\n\t\t\t\t\t<artifactId>jmxtools</artifactId>\r\n\t\t\t\t\t<groupId>com.sun.jdmk</groupId>\r\n\t\t\t\t</exclusion>\r\n\t\t\t\t<exclusion>\r\n\t\t\t\t\t<artifactId>log4j</artifactId>\r\n\t\t\t\t\t<groupId>log4j</groupId>\r\n\t\t\t\t</exclusion>\r\n\t\t\t</exclusions>\r\n\t\t</dependency>\r\n\t\t<dependency>\r\n\t\t\t<groupId>org.springframework.integration</groupId>\r\n\t\t\t<artifactId>spring-integration-stream</artifactId>\r\n\t\t\t<version>${spring.version}</version>\r\n\t\t\t<scope>compile</scope>\r\n\t\t</dependency>\r\n\t\t<dependency>\r\n\t\t\t<groupId>commons-logging</groupId>\r\n\t\t\t<artifactId>commons-logging</artifactId>\r\n\t\t\t<version>1.1.1</version>\r\n\t\t</dependency>\r\n\t\t<dependency>\r\n\t\t\t<groupId>org.slf4j</groupId>\r\n\t\t\t<artifactId>slf4j-log4j12</artifactId>\r\n\t\t\t<version>1.7.5</version>\r\n\t\t</dependency>\r\n\t\t<!-- <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> \r\n\t\t\t<version>0.8.2.1</version> </dependency> -->\r\n\t</dependencies>\r\n\t<build>\r\n\t\t<finalName>\r\n\t\tKafKaWithSpringMVC\r\n\t</finalName>\r\n\t</build>\r\n</project>\r\n\r\nProducer Code:\r\n\r\npackage com;\r\n\r\nimport org.apache.commons.logging.Log;\r\nimport org.apache.commons.logging.LogFactory;\r\nimport org.springframework.beans.factory.annotation.Autowired;\r\nimport org.springframework.integration.kafka.support.KafkaHeaders;\r\nimport org.springframework.integration.support.MessageBuilder;\r\nimport org.springframework.messaging.MessageChannel;\r\nimport org.springframework.stereotype.Controller;\r\nimport org.springframework.ui.Model;\r\nimport org.springframework.web.bind.annotation.RequestMapping;\r\nimport org.springframework.web.bind.annotation.RequestParam;\r\nimport org.springframework.web.servlet.config.annotation.EnableWebMvc;\r\n\r\n@Controller\r\n@EnableWebMvc\r\npublic class ProducerController {\r\n\tprivate static final Log LOG = LogFactory.getLog(ProducerController.class);\r\n\t@Autowired\r\n\tprivate MessageChannel inputToKafka;\r\n\t@RequestMapping(\"/producer\")\r\n\tpublic String sendMessage(@RequestParam(value = \"name\", required = false, defaultValue = \"World\") String name,\r\n\t\t\tModel model) {\r\n\t\tmodel.addAttribute(\"name\", name);\r\n\t\tfor (int i = 30; i < 53; i++) {\r\n\t\t\tboolean flag = inputToKafka.send(\r\n\t\t\t\t\tMessageBuilder.withPayload(\"Message: \" + i).setHeader(KafkaHeaders.TOPIC, \"testtopic\").\r\n\t\t\t\t\tsetHeader(KafkaHeaders.PARTITION_ID, 3).build());\r\n\t\t\tSystem.out.println(\"Message send: \" + flag);\r\n\t\t\tLOG.info(\"message sent \" + i + \"--\");\r\n\t\t}\r\n\t\treturn \"producer\";\r\n\t}\r\n}\r\n\r\n\r\nConsumer Code:\r\n\r\npackage com;\r\n\r\nimport java.util.Collection;\r\nimport java.util.Iterator;\r\nimport java.util.List;\r\nimport java.util.Map;\r\nimport java.util.Set;\r\nimport java.util.concurrent.ConcurrentHashMap;\r\n\r\nimport org.apache.commons.logging.Log;\r\nimport org.apache.commons.logging.LogFactory;\r\nimport org.springframework.beans.factory.annotation.Autowired;\r\nimport org.springframework.messaging.MessageChannel;\r\nimport org.springframework.stereotype.Component;\r\n\r\n\r\n@Component\r\npublic class ConsumerController {\r\n\tprivate static final Log LOG = LogFactory.getLog(ConsumerController.class);\r\n\t\r\n\tpublic void processMessage(Map<String, Map<Integer, List<byte[]>>> msgs) {\r\n\t\tfor (Map.Entry<String, Map<Integer, List<byte[]>>> entry : msgs.entrySet()) {\r\n\t\t\tLOG.debug(\"Topic:\" + entry.getKey());\r\n\t\t\tConcurrentHashMap<Integer, List<byte[]>> messages = (ConcurrentHashMap<Integer, List<byte[]>>) entry\r\n\t\t\t\t\t.getValue();\r\n\t\t\tLOG.debug(\"\\n**** Partition: \\n\");\r\n\t\t\tSet<Integer> keys = messages.keySet();\r\n\t\t\tfor (Integer i : keys)\r\n\t\t\t\tLOG.debug(\"p:\" + i);\r\n\t\t\tLOG.debug(\"\\n**************\\n\");\r\n\t\t\tCollection<List<byte[]>> values = messages.values();\r\n\t\t\tfor (Iterator<List<byte[]>> iterator = values.iterator(); iterator.hasNext();) {\r\n\t\t\t\tList<byte[]> list = iterator.next();\r\n\t\t\t\tfor (byte[] object : list) {\r\n\t\t\t\t\tString message = new String(object);\r\n\t\t\t\t\tLOG.debug(\"Message: \" + message);\r\n\t\t\t\t\ttry {\r\n\t\t\t\t\t\tSystem.out.println(\"Message received: \" + message);\r\n\t\t\t\t\t} catch (Exception e) {\r\n\t\t\t\t\t\tLOG.error(String.format(\"Failed to process message %s\", message));\r\n\t\t\t\t\t}\r\n\t\t\t\t}\r\n\t\t\t}\r\n\t\t}\r\n\t}\r\n}\r\n\r\n\r\nKafka Producer configuration:\r\n\r\n<?xml version=\"1.0\" encoding=\"UTF-8\"?>\r\n<beans xmlns=\"http://www.springframework.org/schema/beans\"\r\n\t   xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\"\r\n\t   xmlns:int=\"http://www.springframework.org/schema/integration\"\r\n\t   xmlns:int-kafka=\"http://www.springframework.org/schema/integration/kafka\"\r\n\t   xmlns:task=\"http://www.springframework.org/schema/task\"\r\n\t   xsi:schemaLocation=\"http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd\r\n\t\thttp://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd\r\n\t\thttp://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd\r\n\t\thttp://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd\">\r\n\t<int:channel id=\"inputToKafka\">\r\n\t\t<int:queue />\r\n\t</int:channel>\r\n\t<int-kafka:outbound-channel-adapter\r\n\t\tid=\"kafkaOutboundChannelAdapter\" kafka-producer-context-ref=\"kafkaProducerContext\"\r\n\t\tchannel=\"inputToKafka\" >\r\n\t\t<int:poller fixed-delay=\"1000\" time-unit=\"MILLISECONDS\"\r\n\t\t\treceive-timeout=\"0\" task-executor=\"taskExecutor\" />\r\n\t</int-kafka:outbound-channel-adapter>\r\n\t<task:executor id=\"taskExecutor\" pool-size=\"5\"\r\n\t\tkeep-alive=\"120\" queue-capacity=\"10000\" />\r\n\t<int-kafka:producer-context id=\"kafkaProducerContext\">\r\n\t\t<int-kafka:producer-configurations>\r\n\t\t\t<int-kafka:producer-configuration\r\n\t\t\t\tbroker-list=\"localhost:9092\" key-class-type=\"java.lang.String\"\r\n\t\t\t\tkey-encoder=\"encoder\" value-class-type=\"java.lang.String\" \r\n\t\t\t\tvalue-encoder=\"encoder\" partitioner=\"partitioner\" topic=\"testtopic\"/>\r\n\t\t</int-kafka:producer-configurations>\r\n\t</int-kafka:producer-context>\r\n\t<bean id=\"encoder\"\r\n\t\tclass=\"org.springframework.integration.kafka.serializer.common.StringEncoder\" />\r\n\t<bean id=\"partitioner\"\r\n\t\tclass=\"org.springframework.integration.kafka.support.DefaultPartitioner\" />\r\n</beans>\r\n\r\nKafka Consumer configuration:\r\n\r\n<?xml version=\"1.0\" encoding=\"UTF-8\"?>\r\n<beans xmlns=\"http://www.springframework.org/schema/beans\"\r\n\txmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\" xmlns:int=\"http://www.springframework.org/schema/integration\"\r\n\txmlns:stream=\"http://www.springframework.org/schema/integration/stream\"\r\n\txmlns:int-kafka=\"http://www.springframework.org/schema/integration/kafka\"\r\n\txmlns:task=\"http://www.springframework.org/schema/task\"\r\n\txsi:schemaLocation=\"http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd\r\n\t\thttp://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd\r\n\t\thttp://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd\r\n\t\thttp://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd\r\n\t\thttp://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd\">\r\n\r\n\t<int:channel id=\"inputFromKafka\">\r\n\t\t<int:dispatcher task-executor=\"kafkaMessageExecutor\" />\r\n\t</int:channel>\r\n\t<task:executor id=\"kafkaMessageExecutor\" pool-size=\"0-10\"\r\n\t\tkeep-alive=\"120\" queue-capacity=\"500\" />\r\n\r\n\t<int-kafka:zookeeper-connect id=\"zookeeperConnect\"\r\n\t\tzk-connect=\"localhost:2181\" zk-connection-timeout=\"6000\"\r\n\t\tzk-session-timeout=\"80\" zk-sync-time=\"2000\" />\r\n\r\n\t<int-kafka:inbound-channel-adapter\r\n\t\tid=\"kafkaInboundChannelAdapter\" kafka-consumer-context-ref=\"consumerContext\"\r\n\t\tauto-startup=\"true\" channel=\"inputFromKafka\">\r\n\t\t<int:poller fixed-delay=\"2000\" time-unit=\"MILLISECONDS\" />\r\n\t</int-kafka:inbound-channel-adapter>\r\n\t<bean id=\"consumerProperties\"\r\n\t\tclass=\"org.springframework.beans.factory.config.PropertiesFactoryBean\">\r\n\t\t<property name=\"properties\">\r\n\t\t\t<props>\r\n\t\t\t\t<prop key=\"auto.offset.reset\">smallest</prop>\r\n\t\t\t\t<prop key=\"socket.receive.buffer.bytes\">10485760</prop> <!-- 10M -->\r\n\t\t\t\t<prop key=\"fetch.message.max.bytes\">5242880</prop>\r\n\t\t\t\t<prop key=\"auto.commit.interval.ms\">1000</prop>\r\n\t\t\t</props>\r\n\t\t</property>\r\n\t</bean>\r\n\t<int:outbound-channel-adapter channel=\"inputFromKafka\"\r\n\t\tref=\"consumerController\" method=\"processMessage\" />\r\n\t<int-kafka:consumer-context id=\"consumerContext\"\r\n\t\tconsumer-timeout=\"1000\" zookeeper-connect=\"zookeeperConnect\"\r\n\t\tconsumer-properties=\"consumerProperties\">\r\n\t\t<int-kafka:consumer-configurations>\r\n\t\t\t<int-kafka:consumer-configuration\r\n\t\t\t\tgroup-id=\"default1\" max-messages=\"5000\" key-decoder=\"deccoder\"\r\n\t\t\t\tvalue-decoder=\"deccoder\">\r\n\t\t\t\t<int-kafka:topic id=\"testtopic\" streams=\"3\" />\r\n\t\t\t</int-kafka:consumer-configuration>\r\n\t\t\t<!-- <int-kafka:consumer-configuration group-id=\"default2\" max-messages=\"50\"> \r\n\t\t\t\t<int-kafka:topic id=\"test2\" streams=\"4\"/> </int-kafka:consumer-configuration> \r\n\t\t\t\t<int-kafka:consumer-configuration group-id=\"default3\" max-messages=\"10\"> \r\n\t\t\t\t<int-kafka:topic-filter pattern=\"regextopic.*\" streams=\"4\" exclude=\"false\"/> \r\n\t\t\t\t</int-kafka:consumer-configuration> -->\r\n\t\t</int-kafka:consumer-configurations>\r\n\t</int-kafka:consumer-context>\r\n\r\n\t<bean id=\"deccoder\"\r\n\t\tclass=\"org.springframework.integration.kafka.serializer.common.StringDecoder\" />\r\n</beans>"}}