title: org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
body: <p>I am manually starting Zookeeper, then Kafka server and finally the Kafka-Rest server with their respective properties file. Next, I am deploying my Spring Boot application on tomcat</p>
  
  <p>In the Tomcat log trace, I am getting the Error <code>org.springframework.context.ApplicationContextException: Failed to start bean 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'; nested exception is org.apache.kafka.common.KafkaException: Failed to construct kafka consumer</code> and my application is failing to startup </p>
  
  <blockquote>
    <p>Error Log</p>
  </blockquote>
  
  <pre><code>25-Dec-2017 15:00:32.508 SEVERE [localhost-startStop-1] org.apache.catalina.core.ContainerBase.addChildInternal ContainerBase.addChild: start:
   org.apache.catalina.LifecycleException: Failed to start component [StandardEngine[Catalina].StandardHost[localhost].StandardContext[/spring-kafka-webhook-service-0.0.1-SNAPSHOT]]
          at org.apache.catalina.util.LifecycleBase.start(LifecycleBase.java:167)
          at org.apache.catalina.core.ContainerBase.addChildInternal(ContainerBase.java:752)
          at org.apache.catalina.core.ContainerBase.addChild(ContainerBase.java:728)
          at org.apache.catalina.core.StandardHost.addChild(StandardHost.java:734)
          at org.apache.catalina.startup.HostConfig.deployWAR(HostConfig.java:986)
          at org.apache.catalina.startup.HostConfig$DeployWar.run(HostConfig.java:1857)
          at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
          at java.util.concurrent.FutureTask.run(FutureTask.java:266)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
          at java.lang.Thread.run(Thread.java:748)
  Caused by: org.springframework.context.ApplicationContextException: Failed to start bean 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'; nested exception is org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
          at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:178)
          at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:50)
          at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:348)
          at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:151)
          at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:114)
          at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:880)
          at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.finishRefresh(EmbeddedWebApplicationContext.java:144)
          at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:546)
          at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.refresh(EmbeddedWebApplicationContext.java:122)
          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.web.support.SpringBootServletInitializer.run(SpringBootServletInitializer.java:154)
          at org.springframework.boot.web.support.SpringBootServletInitializer.createRootApplicationContext(SpringBootServletInitializer.java:134)
          at org.springframework.boot.web.support.SpringBootServletInitializer.onStartup(SpringBootServletInitializer.java:87)
          at org.springframework.web.SpringServletContainerInitializer.onStartup(SpringServletContainerInitializer.java:169)
          at org.apache.catalina.core.StandardContext.startInternal(StandardContext.java:5196)
          at org.apache.catalina.util.LifecycleBase.start(LifecycleBase.java:150)
          ... 10 more
  Caused by: org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
          at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:702)
          at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:557)
          at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createKafkaConsumer(DefaultKafkaConsumerFactory.java:73)
          at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createConsumer(DefaultKafkaConsumerFactory.java:69)
          at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.&lt;init&gt;(KafkaMessageListenerContainer.java:305)
          at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.&lt;init&gt;(KafkaMessageListenerContainer.java:230)
          at org.springframework.kafka.listener.KafkaMessageListenerContainer.doStart(KafkaMessageListenerContainer.java:180)
          at org.springframework.kafka.listener.AbstractMessageListenerContainer.start(AbstractMessageListenerContainer.java:202)
          at org.springframework.kafka.listener.ConcurrentMessageListenerContainer.doStart(ConcurrentMessageListenerContainer.java:126)
          at org.springframework.kafka.listener.AbstractMessageListenerContainer.start(AbstractMessageListenerContainer.java:202)
          at org.springframework.kafka.config.KafkaListenerEndpointRegistry.startIfNecessary(KafkaListenerEndpointRegistry.java:287)
          at org.springframework.kafka.config.KafkaListenerEndpointRegistry.start(KafkaListenerEndpointRegistry.java:236)
          at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:175)
          ... 27 more
  Caused by: java.lang.NoClassDefFoundError: org/apache/kafka/common/ClusterResourceListener
          at java.lang.ClassLoader.defineClass1(Native Method)
          at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
          at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
          at org.apache.catalina.loader.WebappClassLoaderBase.findClassInternal(WebappClassLoaderBase.java:2283)
          at org.apache.catalina.loader.WebappClassLoaderBase.findClass(WebappClassLoaderBase.java:811)
          at org.apache.catalina.loader.WebappClassLoaderBase.loadClass(WebappClassLoaderBase.java:1260)
          at org.apache.catalina.loader.WebappClassLoaderBase.loadClass(WebappClassLoaderBase.java:1119)
          at java.lang.Class.forName0(Native Method)
          at java.lang.Class.forName(Class.java:348)
          at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:332)
          at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstances(AbstractConfig.java:225)
          at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:643)
          ... 39 more
  Caused by: java.lang.ClassNotFoundException: org.apache.kafka.common.ClusterResourceListener
          at org.apache.catalina.loader.WebappClassLoaderBase.loadClass(WebappClassLoaderBase.java:1291)
          at org.apache.catalina.loader.WebappClassLoaderBase.loadClass(WebappClassLoaderBase.java:1119)
          ... 51 more
  </code></pre>
  
  <blockquote>
    <p>Receiver class </p>
  </blockquote>
  
  <pre><code>public class InventoryEventReceiver {
  
      private static final Logger log = LoggerFactory.getLogger(InventoryEventReceiver.class);
  
      private CountDownLatch latch = new CountDownLatch(1);
  
      public CountDownLatch getLatch() {
          return latch;
      }
  
      @KafkaListener(topics="inventory", containerFactory="kafkaListenerContainerFactory")
      public void listenWithHeaders(
              @Payload InventoryEvent event,
              @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
              @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) Integer key,
              @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
              @Header(KafkaHeaders.OFFSET) String offset
              ) {
  
          System.out.println("EVENT HAS BEEN RECEIVED by listenWithHeaders(InventoryEvent)");
          System.out.println(event.toString());
  
  
          log.info(System.currentTimeMillis() + "-- Received Event :\"" + event + "\" from partition:offset -- " + partition + ":" + offset +
                  " for topic : " + topic);       
  
          String urlForInventoryListeners = "http://localhost:8080/" + topic + "/listeners";
          OutputStream os = null;
          try {
              URL objectUrl = new URL(urlForInventoryListeners);
              HttpURLConnection con = (HttpURLConnection) objectUrl.openConnection();
              con.setRequestMethod("POST");
              con.setRequestProperty("Content-Type", "application/json; charset=UTF-8");
              con.setRequestProperty("topic", topic);
              Gson gson = new Gson();
              String eventJson = gson.toJson(event);
              con.setDoOutput(true);
              os = con.getOutputStream();
              os.write(eventJson.getBytes("UTF-8"));
              System.out.println("Event sent to " + objectUrl);
  
          } catch (Exception e) {
              e.printStackTrace();
              System.out.println(e.getMessage());
          } finally {
              try {
                  os.close();
              } catch (IOException e) {
                  e.printStackTrace();
                  System.out.println(e.getMessage());
              }
          }
  
          latch.countDown();
      }
  
  }
  </code></pre>
  
  <blockquote>
    <p>Receiver config class</p>
  </blockquote>
  
  <pre><code>@Configuration
  @EnableKafka
  public class InventoryReceiverConfig {
  
      @Autowired
      private KafkaConfig kafkaConfig;
  
      @Bean
      public static ConsumerFactory&lt;String, InventoryEvent&gt; consumerFactory() { 
          return new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs(), new StringDeserializer(), 
                  new JsonDeserializer&lt;&gt;(InventoryEvent.class));
      }
  
      @Bean
      public static ConcurrentKafkaListenerContainerFactory&lt;String, InventoryEvent&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, InventoryEvent&gt; containerFactory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          containerFactory.setConsumerFactory(consumerFactory());
          containerFactory.setConcurrency(3); 
          containerFactory.getContainerProperties().setPollTimeout(3000);
          return containerFactory;
      }
  
      @Bean
      public static Map&lt;String, Object&gt; consumerConfigs() {
          Map&lt;String, Object&gt; consumerProps = new HashMap&lt;&gt;();
          consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
          consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG,"inventory_consumers");
          consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
          consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,JsonDeserializer.class);
          consumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor");
          return consumerProps;
      }
  
      @Bean
      public InventoryEventReceiver receiver() {
          return new InventoryEventReceiver();
      }
  
  }
  </code></pre>
  
  <p>And my cluster properties file for server.properties, consumer.properties and kafka-rest.properties are as follows:</p>
  
  <blockquote>
    <p>server.properties</p>
  </blockquote>
  
  <pre><code># The id of the broker. This must be set to a unique integer for each broker.
  broker.id=0
  
  # Switch to enable topic deletion or not, default value is false
  delete.topic.enable=true
  
  ############################# Socket Server Settings #############################
  
  # The address the socket server listens on. It will get the value returned from
  # java.net.InetAddress.getCanonicalHostName() if not configured.
  #   FORMAT:
  #     listeners = listener_name://host_name:port
  #   EXAMPLE:
  #     listeners = PLAINTEXT://your.host.name:9092
  listeners=PLAINTEXT://:9092
  
  # Hostname and port the broker will advertise to producers and consumers. If not set,
  # it uses the value for "listeners" if configured.  Otherwise, it will use the value
  # returned from java.net.InetAddress.getCanonicalHostName().
  advertised.listeners=PLAINTEXT://localhost:9092
  # The number of threads that the server uses for receiving requests from the network and sending responses to the network
  num.network.threads=3
  
  # The number of threads that the server uses for processing requests, which may include disk I/O
  num.io.threads=8
  
  # The send buffer (SO_SNDBUF) used by the socket server
  socket.send.buffer.bytes=102400
  
  # The receive buffer (SO_RCVBUF) used by the socket server
  socket.receive.buffer.bytes=102400
  
  # The maximum size of a request that the socket server will accept (protection against OOM)
  socket.request.max.bytes=104857600
  # A comma seperated list of directories under which to store log files
  log.dirs=/tmp/kafka-logs
  
  # The default number of log partitions per topic. More partitions allow greater
  # parallelism for consumption, but this will also result in more files across
  # the brokers.
  num.partitions=1
  
  # The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
  # This value is recommended to be increased for installations with data dirs located in RAID array.
  num.recovery.threads.per.data.dir=1
  
  ############################# Internal Topic Settings  #############################
  # The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
  # For anything other than development testing, a value greater than 1 is recommended for to ensure availability such as 3.
  offsets.topic.replication.factor=1
  transaction.state.log.replication.factor=1
  transaction.state.log.min.isr=1
  # The minimum age of a log file to be eligible for deletion due to age
  log.retention.hours=168
  # The maximum size of a log segment file. When this size is reached a new log segment will be created.
  log.segment.bytes=1073741824
  
  # The interval at which log segments are checked to see if they can be deleted according
  # to the retention policies
  log.retention.check.interval.ms=300000
  # Zookeeper connection string (see zookeeper docs for details).
  # This is a comma separated host:port pairs, each corresponding to a zk
  # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
  # You can also append an optional chroot string to the urls to specify the
  # root directory for all kafka znodes.
  zookeeper.connect=localhost:2181
  
  # Timeout in ms for connecting to zookeeper
  zookeeper.connection.timeout.ms=6000
  ##################### Confluent Proactive Support ######################
  # If set to true, and confluent-support-metrics package is installed
  # then the feature to collect and report support metrics
  # ("Metrics") is enabled.  If set to false, the feature is disabled.
  #
  confluent.support.metrics.enable=true
  ############################# Group Coordinator Settings #############################
  
  # The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
  # The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
  # The default value for this is 3 seconds.
  # We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
  # However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
  group.initial.rebalance.delay.ms=0
  
  
  # The customer ID under which support metrics will be collected and
  # reported.
  #
  # When the customer ID is set to "anonymous" (the default), then only a
  # reduced set of metrics is being collected and reported.
  #
  # Confluent customers
  # -------------------
  # If you are a Confluent customer, then you should replace the default
  # value with your actual Confluent customer ID.  Doing so will ensure
  # that additional support metrics will be collected and reported.
  #
  confluent.support.customer.id=anonymous
  </code></pre>
  
  <blockquote>
    <p>consumer.properties</p>
  </blockquote>
  
  <pre><code># Zookeeper connection string
  # comma separated host:port pairs, each corresponding to a zk
  # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002"
  zookeeper.connect=127.0.0.1:2181
  
  # timeout in ms for connecting to zookeeper
  zookeeper.connection.timeout.ms=6000
  
  #consumer group id
  group.id=test-consumer-group,inventory_consumers
  
  #consumer timeout
  #consumer.timeout.ms=5000
  </code></pre>
  
  <blockquote>
    <p>kafka-rest.properties</p>
  </blockquote>
  
  <pre><code>id=kafka-rest-test-server
  schema.registry.url=http://localhost:8081
  zookeeper.connect=localhost:2181
  #
  # Configure interceptor classes for sending consumer and producer metrics to Confluent Control Center
  # Make sure that monitoring-interceptors-&lt;version&gt;.jar is on the Java class path
  consumer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor
  producer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor
  </code></pre>
  
  <blockquote>
    <p>pom.xml</p>
  </blockquote>
  
  <pre><code>&lt;?xml version="1.0" encoding="UTF-8"?&gt;
  &lt;project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"&gt;
      &lt;modelVersion&gt;4.0.0&lt;/modelVersion&gt;
  
      &lt;groupId&gt;com.psl.kafka.spring&lt;/groupId&gt;
      &lt;artifactId&gt;spring-kafka-webhook-service&lt;/artifactId&gt;
      &lt;packaging&gt;war&lt;/packaging&gt;
  
      &lt;name&gt;spring-kafka-webhook-service&lt;/name&gt;
      &lt;description&gt;Spring Kafka Webhook Service&lt;/description&gt;
  
      &lt;parent&gt;
          &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
          &lt;artifactId&gt;spring-boot-starter-parent&lt;/artifactId&gt;
          &lt;version&gt;1.5.9.RELEASE&lt;/version&gt;
          &lt;relativePath /&gt; &lt;!-- lookup parent from repository --&gt;
      &lt;/parent&gt;
  
      &lt;repositories&gt;
          &lt;repository&gt;
              &lt;id&gt;confluent&lt;/id&gt;
              &lt;url&gt;http://packages.confluent.io/maven/&lt;/url&gt;
          &lt;/repository&gt;
      &lt;/repositories&gt;
  
      &lt;properties&gt;
          &lt;project.build.sourceEncoding&gt;UTF-8&lt;/project.build.sourceEncoding&gt;
          &lt;project.reporting.outputEncoding&gt;UTF-8&lt;/project.reporting.outputEncoding&gt;
          &lt;java.version&gt;1.8&lt;/java.version&gt;
          &lt;org.springframework-version&gt;5.0.0.RELEASE&lt;/org.springframework-version&gt;
          &lt;org.springframework.security-version&gt;4.0.1.RELEASE&lt;/org.springframework.security-version&gt;
          &lt;org.aspectj-version&gt;1.8.11&lt;/org.aspectj-version&gt;
          &lt;org.slf4j-version&gt;1.7.12&lt;/org.slf4j-version&gt;
      &lt;/properties&gt;
  
      &lt;dependencies&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
              &lt;artifactId&gt;spring-boot-starter-web&lt;/artifactId&gt;
          &lt;/dependency&gt;
  
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
              &lt;artifactId&gt;spring-boot-starter-tomcat&lt;/artifactId&gt;
              &lt;scope&gt;provided&lt;/scope&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
              &lt;artifactId&gt;spring-boot-starter-test&lt;/artifactId&gt;
              &lt;scope&gt;test&lt;/scope&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.kafka&lt;/groupId&gt;
              &lt;artifactId&gt;spring-kafka&lt;/artifactId&gt;
              &lt;version&gt;1.1.1.RELEASE&lt;/version&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.kafka&lt;/groupId&gt;
              &lt;artifactId&gt;spring-kafka-test&lt;/artifactId&gt;
              &lt;version&gt;1.1.1.RELEASE&lt;/version&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;log4j&lt;/groupId&gt;
              &lt;artifactId&gt;log4j&lt;/artifactId&gt;
              &lt;version&gt;1.2.16&lt;/version&gt;
              &lt;scope&gt;runtime&lt;/scope&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;com.fasterxml.jackson.core&lt;/groupId&gt;
              &lt;artifactId&gt;jackson-databind&lt;/artifactId&gt;
              &lt;version&gt;2.9.2&lt;/version&gt;
              &lt;exclusions&gt;
                  &lt;exclusion&gt;
                      &lt;groupId&gt;com.fasterxml.jackson.core&lt;/groupId&gt;
                      &lt;artifactId&gt;jackson-core&lt;/artifactId&gt;
                  &lt;/exclusion&gt;
                  &lt;exclusion&gt;
                      &lt;groupId&gt;com.fasterxml.jackson.core&lt;/groupId&gt;
                      &lt;artifactId&gt;jackson-annotations&lt;/artifactId&gt;
                  &lt;/exclusion&gt;
              &lt;/exclusions&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;com.fasterxml.jackson.core&lt;/groupId&gt;
              &lt;artifactId&gt;jackson-core&lt;/artifactId&gt;
              &lt;version&gt;2.9.2&lt;/version&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;com.fasterxml.jackson.core&lt;/groupId&gt;
              &lt;artifactId&gt;jackson-annotations&lt;/artifactId&gt;
              &lt;version&gt;2.9.2&lt;/version&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.json&lt;/groupId&gt;
              &lt;artifactId&gt;json&lt;/artifactId&gt;
              &lt;version&gt;20160810&lt;/version&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;com.google.code.gson&lt;/groupId&gt;
              &lt;artifactId&gt;gson&lt;/artifactId&gt;
              &lt;version&gt;2.8.0&lt;/version&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.integration&lt;/groupId&gt;
              &lt;artifactId&gt;spring-integration-kafka&lt;/artifactId&gt;
              &lt;version&gt;2.1.0.RELEASE&lt;/version&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.apache.kafka&lt;/groupId&gt;
              &lt;artifactId&gt;kafka-clients&lt;/artifactId&gt;
              &lt;version&gt;0.10.0.1&lt;/version&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;io.confluent&lt;/groupId&gt;
              &lt;artifactId&gt;monitoring-interceptors&lt;/artifactId&gt;
              &lt;version&gt;3.1.1&lt;/version&gt;
          &lt;/dependency&gt;
      &lt;/dependencies&gt;
  
      &lt;build&gt;
          &lt;plugins&gt;
              &lt;plugin&gt;
                  &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
                  &lt;artifactId&gt;spring-boot-maven-plugin&lt;/artifactId&gt;
              &lt;/plugin&gt;
          &lt;/plugins&gt;
      &lt;/build&gt;
  
  
      &lt;version&gt;0.0.1-SNAPSHOT&lt;/version&gt;
  &lt;/project&gt;
  </code></pre>
  
  <p>My Receiver and Sender classes are not annotated with any annotations such as <code>@Component</code> or <code>@Service</code>. Does it make any difference?</p>
  
  <pre><code>@Configuration 
  public class InventorySenderConfig
  
  @Configuration 
  @EnableKafka 
  public class InventoryReceiverConfig
  
  @Component 
  public class KafkaConfig
  
  @Configuration 
  public class ProducingChannelConfig
  
  @Configuration 
  public class ConsumingChannelConfig
  
  @RestController 
  public class KafkaWebhookController
  
  @Service("webhookService") 
  public class KafkaServiceImpl
  
  @EnableIntegration 
  @SpringBootApplication 
  @ComponentScan("com.psl.kafka") 
  public class SpringKafkaWebhookServiceApplication extends SpringBootServletInitializer
  </code></pre>
  
  <p>These are my class annotations. Do they look to be OK or I need to change something ?</p>
  
  <blockquote>
    <p>New Build Error after kafka version update to 0.10.1.1</p>
  </blockquote>
  
  <pre><code>2017-12-26 13:11:44.490  INFO 13444 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 0.10.1.1
  2017-12-26 13:11:44.490  INFO 13444 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : f10ef2720b03b247
  2017-12-26 13:12:44.499 ERROR 13444 --- [           main] o.s.k.support.LoggingProducerListener    : Exception thrown when sending a message with key='inventory-events' and payload='Hello Spring Integration Kafka 0!' to topic inventory:
  
  org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
  
  2017-12-26 13:12:44.501  WARN 13444 --- [           main] o.a.k.c.p.i.ProducerInterceptors         : Error executing interceptor onAcknowledgement callback
  
  java.lang.IllegalStateException: clusterResource is not defined
      at io.confluent.shaded.com.google.common.base.Preconditions.checkState(Preconditions.java:174) ~[monitoring-interceptors-3.1.1.jar:na]
      at io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor.onAcknowledgement(MonitoringProducerInterceptor.java:59) ~[monitoring-interceptors-3.1.1.jar:na]
      at org.apache.kafka.clients.producer.internals.ProducerInterceptors.onSendError(ProducerInterceptors.java:116) ~[kafka-clients-0.10.1.1.jar:na]
      at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:489) [kafka-clients-0.10.1.1.jar:na]
      at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:436) [kafka-clients-0.10.1.1.jar:na]
  </code></pre>
  
  <p>Do I need to define any Interceptor classes that I have added as config in ProducingChannelConfig, ConsumingChannelConfig as well as InventoryReceiverConfig classes ?</p>