GH-13: Add checkstype and fix as much as possible

Fixes GH-13 (https://github.com/spring-projects/spring-kafka/issues/13)

Also upgrade to Gradle 2.11 to support the latest `checkstyle`

build.gradle

@@ -47,6 +47,7 @@ subprojects { subproject ->
 	apply plugin: 'eclipse'
 	apply plugin: 'idea'
 	apply plugin: 'jacoco'
+	apply plugin: 'checkstyle'
 
 	if (project.hasProperty('platformVersion')) {
 		apply plugin: 'spring-io'
@@ -100,6 +101,11 @@ subprojects { subproject ->
 		}
 	}
 
+	checkstyle {
+		configFile = new File(rootDir, "src/checkstyle/checkstyle.xml")
+		toolVersion = "6.16.1"
+	}
+
 	jacocoTestReport {
 		reports {
 			xml.enabled false
@@ -325,9 +331,3 @@ task dist(dependsOn: assemble) {
 	group = 'Distribution'
 	description = 'Builds -dist, -docs distribution archives.'
 }
-
-task wrapper(type: Wrapper) {
-	description = 'Generates gradlew[.bat] scripts'
-	gradleVersion = '2.5'
-	distributionUrl = "http://services.gradle.org/distributions/gradle-${gradleVersion}-all.zip"
-}

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
-#Wed Sep 02 11:48:49 EDT 2015
+#Mon Mar 07 20:47:12 EST 2016
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=http\://services.gradle.org/distributions/gradle-2.5-all.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-2.11-bin.zip

gradlew.bat

@@ -46,7 +46,7 @@ echo location of your Java installation.
 goto fail
 
 :init
-@rem Get command-line arguments, handling Windowz variants
+@rem Get command-line arguments, handling Windows variants
 
 if not "%OS%" == "Windows_NT" goto win9xME_args
 if "%@eval[2+2]" == "4" goto 4NT_args

spring-kafka-test/src/main/java/org/springframework/kafka/core/BrokerAddress.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -14,7 +14,6 @@
  * limitations under the License.
  */
 
-
 package org.springframework.kafka.core;
 
 import org.springframework.util.Assert;

spring-kafka-test/src/main/java/org/springframework/kafka/rule/KafkaEmbedded.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -14,7 +14,6 @@
  * limitations under the License.
  */
 
-
 package org.springframework.kafka.rule;
 
 import java.io.File;
@@ -116,10 +115,10 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule {
 		int zkSessionTimeout = 6000;
 
 		this.zkConnect = "127.0.0.1:" + this.zookeeper.port();
-		zookeeperClient = new ZkClient(zkConnect, zkSessionTimeout, zkConnectionTimeout,
+		this.zookeeperClient = new ZkClient(this.zkConnect, zkSessionTimeout, zkConnectionTimeout,
 				ZKStringSerializer$.MODULE$);
-		kafkaServers = new ArrayList<KafkaServer>();
-		for (int i = 0; i < count; i++) {
+		this.kafkaServers = new ArrayList<>();
+		for (int i = 0; i < this.count; i++) {
 			ServerSocket ss = ServerSocketFactory.getDefault().createServerSocket(0);
 			int randomPort = ss.getLocalPort();
 			ss.close();
@@ -128,22 +127,22 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule {
 					scala.Option.<SecurityProtocol>apply(null),
 					scala.Option.<File>apply(null),
 					true, false, 0, false, 0, false, 0);
-			brokerConfigProperties.setProperty("replica.socket.timeout.ms","1000");
-			brokerConfigProperties.setProperty("controller.socket.timeout.ms","1000");
-			brokerConfigProperties.setProperty("offsets.topic.replication.factor","1");
+			brokerConfigProperties.setProperty("replica.socket.timeout.ms", "1000");
+			brokerConfigProperties.setProperty("controller.socket.timeout.ms", "1000");
+			brokerConfigProperties.setProperty("offsets.topic.replication.factor", "1");
 			KafkaServer server = TestUtils.createServer(new KafkaConfig(brokerConfigProperties), SystemTime$.MODULE$);
-			kafkaServers.add(server);
+			this.kafkaServers.add(server);
 		}
 		ZkUtils zkUtils = new ZkUtils(getZkClient(), null, false);
 		Properties props = new Properties();
-		for (String topic : topics) {
+		for (String topic : this.topics) {
 			AdminUtils.createTopic(zkUtils, topic, this.partitionsPerTopic, this.count, props);
 		}
 	}
 
 	@Override
 	protected void after() {
-		for (KafkaServer kafkaServer : kafkaServers) {
+		for (KafkaServer kafkaServer : this.kafkaServers) {
 			try {
 				if (kafkaServer.brokerState().currentState() != (NotRunning.state())) {
 					kafkaServer.shutdown();
@@ -161,13 +160,13 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule {
 			}
 		}
 		try {
-			zookeeperClient.close();
+			this.zookeeperClient.close();
 		}
 		catch (ZkInterruptedException e) {
 			// do nothing
 		}
 		try {
-			zookeeper.shutdown();
+			this.zookeeper.shutdown();
 		}
 		catch (Exception e) {
 			// do nothing
@@ -176,25 +175,25 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule {
 
 	@Override
 	public List<KafkaServer> getKafkaServers() {
-		return kafkaServers;
+		return this.kafkaServers;
 	}
 
 	public KafkaServer getKafkaServer(int id) {
-		return kafkaServers.get(id);
+		return this.kafkaServers.get(id);
 	}
 
 	public EmbeddedZookeeper getZookeeper() {
-		return zookeeper;
+		return this.zookeeper;
 	}
 
 	@Override
 	public ZkClient getZkClient() {
-		return zookeeperClient;
+		return this.zookeeperClient;
 	}
 
 	@Override
 	public String getZookeeperConnectionString() {
-		return zkConnect;
+		return this.zkConnect;
 	}
 
 	public BrokerAddress getBrokerAddress(int i) {
@@ -231,11 +230,11 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule {
 	}
 
 	public void startZookeeper() {
-		zookeeper = new EmbeddedZookeeper();
+		this.zookeeper = new EmbeddedZookeeper();
 	}
 
 	public void bounce(int index, boolean waitForPropagation) {
-		kafkaServers.get(index).shutdown();
+		this.kafkaServers.get(index).shutdown();
 		if (waitForPropagation) {
 			long initialTime = System.currentTimeMillis();
 			boolean canExit = false;
@@ -255,7 +254,8 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule {
 					if (Errors.forCode(topicMetadata.errorCode()).exception() == null) {
 						for (PartitionMetadata partitionMetadata :
 								JavaConversions.asJavaCollection(topicMetadata.partitionsMetadata())) {
-							Collection<BrokerEndPoint> inSyncReplicas = JavaConversions.asJavaCollection(partitionMetadata.isr());
+							Collection<BrokerEndPoint> inSyncReplicas =
+									JavaConversions.asJavaCollection(partitionMetadata.isr());
 							for (BrokerEndPoint broker : inSyncReplicas) {
 								if (broker.id() == index) {
 									canExit = false;
@@ -279,7 +279,7 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule {
 		// retry restarting repeatedly, first attempts may fail
 
 		SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(10,
-				Collections.<Class<? extends Throwable>,Boolean>singletonMap(Exception.class, true));
+				Collections.<Class<? extends Throwable>, Boolean>singletonMap(Exception.class, true));
 
 		ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
 		backOffPolicy.setInitialInterval(100);
@@ -292,9 +292,10 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule {
 
 
 		retryTemplate.execute(new RetryCallback<Void, Exception>() {
+
 			@Override
 			public Void doWithRetry(RetryContext context) throws Exception {
-				kafkaServers.get(index).startup();
+				KafkaEmbedded.this.kafkaServers.get(index).startup();
 				return null;
 			}
 		});

spring-kafka-test/src/main/java/org/springframework/kafka/rule/KafkaRule.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -14,7 +14,6 @@
  * limitations under the License.
  */
 
-
 package org.springframework.kafka.rule;
 
 import java.util.List;

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -343,7 +343,8 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 			}
 			catch (NoSuchBeanDefinitionException ex) {
 				throw new BeanInitializationException("Could not register Kafka listener endpoint on [" +
-						adminTarget + "] for bean " + beanName + ", no " + KafkaListenerContainerFactory.class.getSimpleName() + " with id '" +
+						adminTarget + "] for bean " + beanName + ", no "
+						+ KafkaListenerContainerFactory.class.getSimpleName() + " with id '" +
 						containerFactoryBeanName + "' was found in the application context", ex);
 			}
 		}
@@ -356,7 +357,7 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 			return resolve(KafkaListener.id());
 		}
 		else {
-			return "org.springframework.kafka.KafkaListenerEndpointContainer#" + counter.getAndIncrement();
+			return "org.springframework.kafka.KafkaListenerEndpointContainer#" + this.counter.getAndIncrement();
 		}
 	}
 
@@ -414,17 +415,16 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 
 	@SuppressWarnings("unchecked")
 	private void resolveAsString(Object resolvedValue, List<String> result) {
-		Object resolvedValueToUse = resolvedValue;
 		if (resolvedValue instanceof String[]) {
 			for (Object object : (String[]) resolvedValue) {
 				resolveAsString(object, result);
 			}
 		}
-		if (resolvedValueToUse instanceof String) {
-			result.add((String) resolvedValueToUse);
+		if (resolvedValue instanceof String) {
+			result.add((String) resolvedValue);
 		}
-		else if (resolvedValueToUse instanceof Iterable) {
-			for (Object object : (Iterable<Object>) resolvedValueToUse) {
+		else if (resolvedValue instanceof Iterable) {
+			for (Object object : (Iterable<Object>) resolvedValue) {
 				resolveAsString(object, result);
 			}
 		}
@@ -484,7 +484,7 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 
 		private MessageHandlerMethodFactory createDefaultMessageHandlerMethodFactory() {
 			DefaultMessageHandlerMethodFactory defaultFactory = new DefaultMessageHandlerMethodFactory();
-			defaultFactory.setBeanFactory(beanFactory);
+			defaultFactory.setBeanFactory(KafkaListenerAnnotationBeanPostProcessor.this.beanFactory);
 			defaultFactory.afterPropertiesSet();
 			return defaultFactory;
 		}

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListeners.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.annotation;
 
 import java.lang.annotation.Documented;

spring-kafka/src/main/java/org/springframework/kafka/annotation/TopicPartition.java

@@ -13,11 +13,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.annotation;
 
-import static java.lang.annotation.RetentionPolicy.RUNTIME;
-
 import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
 /**
@@ -27,7 +27,7 @@ import java.lang.annotation.Target;
  *
  */
 @Target({})
-@Retention(RUNTIME)
+@Retention(RetentionPolicy.RUNTIME)
 public @interface TopicPartition {
 
 	String topic() default "";

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java

@@ -60,7 +60,7 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 	}
 
 	public ConsumerFactory<K, V> getConsumerFactory() {
-		return consumerFactory;
+		return this.consumerFactory;
 	}
 
 	/**

spring-kafka/src/main/java/org/springframework/kafka/core/ConsumerFactory.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.core;
 
 import org.apache.kafka.clients.consumer.Consumer;

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.core;
 
 import java.util.HashMap;

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.core;
 
 import java.util.HashMap;

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaException.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.core;
 
 /**

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -55,7 +55,7 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 	 * @return the topic.
 	 */
 	public String getDefaultTopic() {
-		return defaultTopic;
+		return this.defaultTopic;
 	}
 
 	/**
@@ -113,12 +113,12 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 				}
 			}
 		}
-		if (logger.isTraceEnabled()) {
-			logger.trace("Sending: " + producerRecord);
+		if (this.logger.isTraceEnabled()) {
+			this.logger.trace("Sending: " + producerRecord);
 		}
 		Future<RecordMetadata> future = this.producer.send(producerRecord);
-		if (logger.isTraceEnabled()) {
-			logger.trace("Sent: " + producerRecord);
+		if (this.logger.isTraceEnabled()) {
+			this.logger.trace("Sent: " + producerRecord);
 		}
 		return future;
 	}

spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.core;
 
 import org.apache.kafka.clients.producer.Producer;

spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractKafkaListenerEndpoint.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2015 the original author or authors.
+ * Copyright 2014-2016 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -72,15 +72,15 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	}
 
 	protected BeanFactory getBeanFactory() {
-		return beanFactory;
+		return this.beanFactory;
 	}
 
 	protected BeanExpressionResolver getResolver() {
-		return resolver;
+		return this.resolver;
 	}
 
 	protected BeanExpressionContext getBeanExpressionContext() {
-		return expressionContext;
+		return this.expressionContext;
 	}
 
 	public void setId(String id) {

spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.listener;
 
 import java.util.concurrent.Executor;
@@ -127,7 +128,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 	}
 
 	public Object getMessageListener() {
-		return messageListener;
+		return this.messageListener;
 	}
 
 	@Override
@@ -159,7 +160,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 	 * @see #setAckMode(AckMode)
 	 */
 	public AckMode getAckMode() {
-		return ackMode;
+		return this.ackMode;
 	}
 
 	/**
@@ -175,7 +176,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 	 * @see #setPollTimeout(long)
 	 */
 	public long getPollTimeout() {
-		return pollTimeout;
+		return this.pollTimeout;
 	}
 
 	/**

spring-kafka/src/main/java/org/springframework/kafka/listener/AcknowledgingMessageListener.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,

spring-kafka/src/main/java/org/springframework/kafka/listener/Acknowledgment.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,

spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -65,12 +65,14 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 	 * @param consumerFactory the consumer factory.
 	 * @param topicPartitions the topics/partitions; duplicates are eliminated.
 	 */
-	public ConcurrentMessageListenerContainer(ConsumerFactory<K, V> consumerFactory, TopicPartition... topicPartitions) {
+	public ConcurrentMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
+	                                          TopicPartition... topicPartitions) {
 		Assert.notNull(consumerFactory, "A ConsumerFactory must be provided");
 		Assert.notEmpty(topicPartitions, "A list of partitions must be provided");
 		Assert.noNullElements(topicPartitions, "The list of partitions cannot contain null elements");
 		this.consumerFactory = consumerFactory;
-		this.partitions = new LinkedHashSet<>(Arrays.asList(topicPartitions)).toArray(new TopicPartition[0]);
+		this.partitions = new LinkedHashSet<>(Arrays.asList(topicPartitions))
+				.toArray(new TopicPartition[topicPartitions.length]);
 		this.topics = null;
 		this.topicPattern = null;
 	}
@@ -131,7 +133,7 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 	}
 
 	public int getConcurrency() {
-		return concurrency;
+		return this.concurrency;
 	}
 
 	/**
@@ -207,7 +209,7 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 				int perContainer = numPartitions / this.concurrency;
 				TopicPartition[] subset;
 				if (i == this.concurrency - 1) {
-					subset = Arrays.copyOfRange(this.partitions, i * perContainer, partitions.length);
+					subset = Arrays.copyOfRange(this.partitions, i * perContainer, this.partitions.length);
 				}
 				else {
 					subset = Arrays.copyOfRange(this.partitions, i * perContainer, (i + 1) * perContainer);

spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandler.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -14,7 +14,6 @@
  * limitations under the License.
  */
 
-
 package org.springframework.kafka.listener;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaListenerEndpointRegistrar.java

@@ -195,7 +195,7 @@ public class KafkaListenerEndpointRegistrar implements BeanFactoryAware, Initial
 	}
 
 
-	private static class KafkaListenerEndpointDescriptor {
+	private static final class KafkaListenerEndpointDescriptor {
 
 		private final KafkaListenerEndpoint endpoint;
 

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaListenerEndpointRegistry.java

@@ -204,7 +204,7 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec
 					((DisposableBean) listenerContainer).destroy();
 				}
 				catch (Exception ex) {
-					logger.warn("Failed to destroy message listener container", ex);
+					this.logger.warn("Failed to destroy message listener container", ex);
 				}
 			}
 		}
@@ -270,7 +270,7 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec
 	}
 
 
-	private static class AggregatingCallback implements Runnable {
+	private static final class AggregatingCallback implements Runnable {
 
 		private final AtomicInteger count;
 

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -117,7 +117,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 	 * @param topicPartitions the topics/partitions; duplicates are eliminated.
 	 */
 	KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory, String[] topics, Pattern topicPattern,
-	                              TopicPartition[] topicPartitions) {
+			TopicPartition[] topicPartitions) {
 		this.consumerFactory = consumerFactory;
 		this.topics = topics;
 		this.topicPattern = topicPattern;
@@ -203,7 +203,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 
 	private class ListenerConsumer implements SchedulingAwareRunnable {
 
-		private final Log logger = LogFactory.getLog(this.getClass());
+		private final Log logger = LogFactory.getLog(ListenerConsumer.class);
 
 		private final CommitCallback callback = new CommitCallback();
 
@@ -221,7 +221,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 
 		private final long recentOffset;
 
-		private final boolean autoCommit = consumerFactory.isAutoCommit();
+		private final boolean autoCommit = KafkaMessageListenerContainer.this.consumerFactory.isAutoCommit();
 
 		private Thread consumerThread;
 
@@ -230,35 +230,35 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 		private volatile Collection<TopicPartition> assignedPartitions;
 
 		public ListenerConsumer(MessageListener<K, V> listener, AcknowledgingMessageListener<K, V> ackListener,
-		                        ContainerOffsetResetStrategy resetStrategy, long recentOffset) {
+				ContainerOffsetResetStrategy resetStrategy, long recentOffset) {
 			Assert.state(!(getAckMode().equals(AckMode.MANUAL) || getAckMode().equals(AckMode.MANUAL_IMMEDIATE))
 					|| !this.autoCommit,
 					"Consumer cannot be configured for auto commit for ackMode " + getAckMode());
-			Consumer<K, V> consumer = consumerFactory.createConsumer();
+			Consumer<K, V> consumer = KafkaMessageListenerContainer.this.consumerFactory.createConsumer();
 			ConsumerRebalanceListener rebalanceListener = new ConsumerRebalanceListener() {
 
 				@Override
 				public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
-					logger.info("partitions revoked:" + partitions);
+					KafkaMessageListenerContainer.this.logger.info("partitions revoked:" + partitions);
 				}
 
 				@Override
 				public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
-					assignedPartitions = partitions;
-					logger.info("partitions assigned:" + partitions);
+					ListenerConsumer.this.assignedPartitions = partitions;
+					KafkaMessageListenerContainer.this.logger.info("partitions assigned:" + partitions);
 				}
 
 			};
-			if (partitions == null) {
-				if (topicPattern != null) {
-					consumer.subscribe(topicPattern, rebalanceListener);
+			if (KafkaMessageListenerContainer.this.partitions == null) {
+				if (KafkaMessageListenerContainer.this.topicPattern != null) {
+					consumer.subscribe(KafkaMessageListenerContainer.this.topicPattern, rebalanceListener);
 				}
 				else {
-					consumer.subscribe(Arrays.asList(topics), rebalanceListener);
+					consumer.subscribe(Arrays.asList(KafkaMessageListenerContainer.this.topics), rebalanceListener);
 				}
 			}
 			else {
-				List<TopicPartition> topicPartitions = Arrays.asList(partitions);
+				List<TopicPartition> topicPartitions = Arrays.asList(KafkaMessageListenerContainer.this.partitions);
 				this.definedPartitions = topicPartitions;
 				consumer.assign(topicPartitions);
 			}
@@ -280,26 +280,26 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			int count = 0;
 			long last = System.currentTimeMillis();
 			long now;
-			if (isRunning() && definedPartitions != null) {
+			if (isRunning() && this.definedPartitions != null) {
 				initPartitionsIfNeeded();
 			}
 			final AckMode ackMode = getAckMode();
 			while (isRunning()) {
 				try {
-					if (logger.isTraceEnabled()) {
-						logger.trace("Polling...");
+					if (this.logger.isTraceEnabled()) {
+						this.logger.trace("Polling...");
 					}
-					ConsumerRecords<K, V> records = consumer.poll(getPollTimeout());
+					ConsumerRecords<K, V> records = this.consumer.poll(getPollTimeout());
 					if (records != null) {
 						count += records.count();
-						if (logger.isDebugEnabled()) {
-							logger.debug("Received: " + records.count() + " records");
+						if (this.logger.isDebugEnabled()) {
+							this.logger.debug("Received: " + records.count() + " records");
 						}
 						Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
 						while (iterator.hasNext()) {
 							final ConsumerRecord<K, V> record = iterator.next();
 							invokeListener(record);
-							if (!autoCommit && ackMode.equals(AckMode.RECORD)) {
+							if (!this.autoCommit && ackMode.equals(AckMode.RECORD)) {
 								this.consumer.commitAsync(
 										Collections.singletonMap(new TopicPartition(record.topic(), record.partition()),
 												new OffsetAndMetadata(record.offset() + 1)), this.callback);
@@ -338,35 +338,35 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 						}
 					}
 					else {
-						if (logger.isDebugEnabled()) {
-							logger.debug("No records");
+						if (this.logger.isDebugEnabled()) {
+							this.logger.debug("No records");
 						}
 					}
 				}
 				catch (WakeupException e) {
-					;
+					// No-op. Continue process
 				}
 				catch (Exception e) {
 					if (getErrorHandler() != null) {
 						getErrorHandler().handle(e, null);
 					}
 					else {
-						logger.error("Container exception", e);
+						this.logger.error("Container exception", e);
 					}
 				}
 			}
-			if (offsets.size() > 0) {
+			if (this.offsets.size() > 0) {
 				commitIfNecessary();
 			}
 			try {
 				this.consumer.unsubscribe();
 			}
 			catch (WakeupException e) {
-				;
+				// No-op. Continue process
 			}
 			this.consumer.close();
-			if (logger.isInfoEnabled()) {
-				logger.info("Consumer stopped");
+			if (this.logger.isInfoEnabled()) {
+				this.logger.info("Consumer stopped");
 			}
 		}
 
@@ -381,18 +381,18 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 								updateManualOffset(record);
 							}
 							else if (getAckMode().equals(AckMode.MANUAL_IMMEDIATE)) {
-								if (Thread.currentThread().equals(consumerThread)) {
+								if (Thread.currentThread().equals(ListenerConsumer.this.consumerThread)) {
 									Map<TopicPartition, OffsetAndMetadata> commits = Collections.singletonMap(
 											new TopicPartition(record.topic(), record.partition()),
 											new OffsetAndMetadata(record.offset() + 1));
-									if (logger.isDebugEnabled()) {
-										logger.debug("Committing: " + commits);
+									if (ListenerConsumer.this.logger.isDebugEnabled()) {
+										ListenerConsumer.this.logger.debug("Committing: " + commits);
 									}
-									consumer.commitAsync(commits, callback);
+									ListenerConsumer.this.consumer.commitAsync(commits, ListenerConsumer.this.callback);
 								}
 								else {
 									throw new IllegalStateException(
-											"With MANUAL_IMMEDIATE ack mode, acknowledget must be invoked on the "
+											"With MANUAL_IMMEDIATE ack mode, acknowledge() must be invoked on the "
 													+ "consumer thread");
 								}
 							}
@@ -405,7 +405,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 					});
 				}
 				else {
-					listener.onMessage(record);
+					this.listener.onMessage(record);
 				}
 			}
 			catch (Exception e) {
@@ -413,7 +413,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 					getErrorHandler().handle(e, record);
 				}
 				else {
-					logger.error("Listener threw an exception and no error handler for " + record, e);
+					this.logger.error("Listener threw an exception and no error handler for " + record, e);
 				}
 			}
 		}
@@ -438,8 +438,8 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 				for (TopicPartition topicPartition : this.definedPartitions) {
 					long newOffset = this.consumer.position(topicPartition) - this.recentOffset;
 					this.consumer.seek(topicPartition, newOffset);
-					if (logger.isDebugEnabled()) {
-						logger.debug("Reset " + topicPartition + " to offset " + newOffset);
+					if (this.logger.isDebugEnabled()) {
+						this.logger.debug("Reset " + topicPartition + " to offset " + newOffset);
 					}
 				}
 			}
@@ -483,8 +483,8 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 				}
 			}
 			this.offsets.clear();
-			if (logger.isDebugEnabled()) {
-				logger.debug("Committing: " + commits);
+			if (this.logger.isDebugEnabled()) {
+				this.logger.debug("Committing: " + commits);
 			}
 			if (!commits.isEmpty()) {
 				this.consumer.commitAsync(commits, this.callback);
@@ -494,7 +494,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 
 	private static final class CommitCallback implements OffsetCommitCallback {
 
-		private final Log logger = LogFactory.getLog(OffsetCommitCallback.class);
+		private static final Log logger = LogFactory.getLog(OffsetCommitCallback.class);
 
 		@Override
 		public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {

spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerExecutionFailedException.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.listener;
 
 import org.springframework.kafka.core.KafkaException;

spring-kafka/src/main/java/org/springframework/kafka/listener/LoggingErrorHandler.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -14,7 +14,6 @@
  * limitations under the License.
  */
 
-
 package org.springframework.kafka.listener;
 
 import org.apache.commons.logging.Log;

spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListener.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -14,7 +14,6 @@
  * limitations under the License.
  */
 
-
 package org.springframework.kafka.listener;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;

spring-kafka/src/main/java/org/springframework/kafka/listener/MethodKafkaListenerEndpoint.java

@@ -79,7 +79,7 @@ public class MethodKafkaListenerEndpoint<K, V> extends AbstractKafkaListenerEndp
 	 * @return the messageHandlerMethodFactory
 	 */
 	protected MessageHandlerMethodFactory getMessageHandlerMethodFactory() {
-		return messageHandlerMethodFactory;
+		return this.messageHandlerMethodFactory;
 	}
 
 	@Override

spring-kafka/src/main/java/org/springframework/kafka/listener/MultiMethodKafkaListenerEndpoint.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.listener;
 
 import java.lang.reflect.Method;

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractAdaptableMessageListener.java

@@ -71,7 +71,7 @@ public abstract class AbstractAdaptableMessageListener<K, V> implements MessageL
 	 * @see #onMessage(ConsumerRecord)
 	 */
 	protected void handleListenerException(Throwable ex) {
-		logger.error("Listener execution failed", ex);
+		this.logger.error("Listener execution failed", ex);
 	}
 
 	/**

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.listener.adapter;
 
 import java.lang.annotation.Annotation;
@@ -60,7 +61,7 @@ public class DelegatingInvocableHandler {
 	 * @return the bean
 	 */
 	public Object getBean() {
-		return bean;
+		return this.bean;
 	}
 
 	/**
@@ -88,7 +89,7 @@ public class DelegatingInvocableHandler {
 			if (handler == null) {
 				throw new KafkaException("No method found for " + payloadClass);
 			}
-			this.cachedHandlers.putIfAbsent(payloadClass, handler);//NOSONAR
+			this.cachedHandlers.putIfAbsent(payloadClass, handler); //NOSONAR
 		}
 		return handler;
 	}
@@ -141,7 +142,7 @@ public class DelegatingInvocableHandler {
 	 */
 	public String getMethodNameFor(Object payload) {
 		InvocableHandlerMethod handlerForPayload = getHandlerForPayload(payload.getClass());
-		return handlerForPayload == null ? "no match" : handlerForPayload.getMethod().toGenericString();//NOSONAR
+		return handlerForPayload == null ? "no match" : handlerForPayload.getMethod().toGenericString(); //NOSONAR
 	}
 
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/HandlerAdapter.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.listener.adapter;
 
 import org.springframework.messaging.Message;

spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessageConverter.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.support.converter;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;

spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java

@@ -56,7 +56,7 @@ public class MessagingMessageConverter<K, V> implements MessageConverter<K, V> {
 
 	@Override
 	public Message<?> toMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment) {
-		KafkaMessageHeaders kafkaMessageHeaders = new KafkaMessageHeaders(generateMessageId, generateTimestamp);
+		KafkaMessageHeaders kafkaMessageHeaders = new KafkaMessageHeaders(this.generateMessageId, this.generateTimestamp);
 
 		Map<String, Object> rawHeaders = kafkaMessageHeaders.getRawHeaders();
 		rawHeaders.put(KafkaHeaders.MESSAGE_KEY, record.key());

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.kafka.annotation;
 
 import static org.junit.Assert.assertEquals;
@@ -116,7 +117,7 @@ public class EnableKafkaIntegrationTests {
 
 		@Bean
 		public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
-							kafkaListenerContainerFactory() {
+		kafkaListenerContainerFactory() {
 			SimpleKafkaListenerContainerFactory<Integer, String> factory = new SimpleKafkaListenerContainerFactory<>();
 			factory.setConsumerFactory(consumerFactory());
 			return factory;
@@ -124,7 +125,7 @@ public class EnableKafkaIntegrationTests {
 
 		@Bean
 		public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
-							kafkaManualAckListenerContainerFactory() {
+		kafkaManualAckListenerContainerFactory() {
 			SimpleKafkaListenerContainerFactory<Integer, String> factory = new SimpleKafkaListenerContainerFactory<>();
 			factory.setConsumerFactory(manualConsumerFactory());
 			factory.setAckMode(AckMode.MANUAL_IMMEDIATE);
@@ -204,24 +205,24 @@ public class EnableKafkaIntegrationTests {
 
 		private volatile Acknowledgment ack;
 
-		@KafkaListener(id="foo", topics = "annotated1")
+		@KafkaListener(id = "foo", topics = "annotated1")
 		public void listen1(String foo) {
 			this.latch1.countDown();
 		}
 
-		@KafkaListener(id="bar", topicPattern = "annotated2")
+		@KafkaListener(id = "bar", topicPattern = "annotated2")
 		public void listen2(@Payload String foo, @Header(KafkaHeaders.PARTITION_ID) int partitionHeader) {
 			this.partition = partitionHeader;
 			this.latch2.countDown();
 		}
 
-		@KafkaListener(id="baz", topicPartitions = @TopicPartition(topic = "annotated3", partition="0"))
+		@KafkaListener(id = "baz", topicPartitions = @TopicPartition(topic = "annotated3", partition = "0"))
 		public void listen3(ConsumerRecord<?, ?> record) {
 			this.record = record;
 			this.latch3.countDown();
 		}
 
-		@KafkaListener(id="qux", topics = "annotated4", containerFactory = "kafkaManualAckListenerContainerFactory")
+		@KafkaListener(id = "qux", topics = "annotated4", containerFactory = "kafkaManualAckListenerContainerFactory")
 		public void listen4(@Payload String foo, Acknowledgment ack) {
 			this.ack = ack;
 			this.ack.acknowledge();

spring-kafka/src/test/java/org/springframework/kafka/core/BrokerAddress.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -14,7 +14,6 @@
  * limitations under the License.
  */
 
-
 package org.springframework.kafka.core;
 
 import org.springframework.util.Assert;

src/checkstyle/checkstyle-header.txt

@@ -0,0 +1,17 @@
+^\Q/*\E$
+^\Q * Copyright \E20\d\d(\-20\d\d)?\Q the original author or authors.\E$
+^\Q *\E$
+^\Q * Licensed under the Apache License, Version 2.0 (the "License");\E$
+^\Q * you may not use this file except in compliance with the License.\E$
+^\Q * You may obtain a copy of the License at\E$
+^\Q *\E$
+^\Q *      http://www.apache.org/licenses/LICENSE-2.0\E$
+^\Q *\E$
+^\Q * Unless required by applicable law or agreed to in writing, software\E$
+^\Q * distributed under the License is distributed on an "AS IS" BASIS,\E$
+^\Q * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\E$
+^\Q * See the License for the specific language governing permissions and\E$
+^\Q * limitations under the License.\E$
+^\Q */\E$
+^$
+^.*$

src/checkstyle/checkstyle-suppressions.xml

@@ -0,0 +1,8 @@
+<?xml version="1.0"?>
+<!DOCTYPE suppressions PUBLIC
+    "-//Puppy Crawl//DTD Suppressions 1.1//EN"
+    "http://www.puppycrawl.com/dtds/suppressions_1_1.dtd">
+<suppressions>
+	<suppress files="package-info\.java" checks=".*" />
+	<suppress files="[\\/]test[\\/]" checks="RequireThis" />
+</suppressions>

src/checkstyle/checkstyle.xml

@@ -0,0 +1,169 @@
+<?xml version="1.0"?>
+<!DOCTYPE module PUBLIC "-//Puppy Crawl//DTD Check Configuration 1.2//EN" "http://www.puppycrawl.com/dtds/configuration_1_2.dtd">
+<module name="Checker">
+
+	<module name="SuppressionFilter">
+		<property name="file" value="src/checkstyle/checkstyle-suppressions.xml" />
+	</module>
+
+	<!-- Root Checks -->
+	<module name="RegexpHeader">
+		<property name="headerFile" value="src/checkstyle/checkstyle-header.txt" />
+		<property name="fileExtensions" value="java" />
+	</module>
+	<module name="NewlineAtEndOfFile">
+		<property name="lineSeparator" value="lf"/>
+	</module>
+
+	<!-- TreeWalker Checks -->
+	<module name="TreeWalker">
+
+		<!-- Annotations -->
+		<module name="AnnotationUseStyle">
+			<property name="elementStyle" value="compact" />
+		</module>
+		<module name="MissingOverride" />
+		<module name="PackageAnnotation" />
+		<module name="AnnotationLocation">
+			<property name="allowSamelineSingleParameterlessAnnotation"
+				value="false" />
+		</module>
+
+		<!-- Block Checks -->
+		<module name="EmptyBlock">
+			<property name="option" value="text" />
+		</module>
+		<module name="LeftCurly" />
+		<module name="RightCurly">
+			<property name="option" value="alone" />
+		</module>
+		<module name="NeedBraces" />
+		<module name="AvoidNestedBlocks" />
+
+		<!-- Class Design -->
+		<module name="FinalClass" />
+		<module name="InterfaceIsType" />
+		<module name="HideUtilityClassConstructor" />
+		<module name="MutableException" />
+		<module name="InnerTypeLast" />
+		<module name="OneTopLevelClass" />
+
+		<!-- Coding -->
+		<module name="CovariantEquals" />
+		<module name="EmptyStatement" />
+		<module name="EqualsHashCode" />
+		<module name="InnerAssignment" />
+		<module name="SimplifyBooleanExpression" />
+		<module name="SimplifyBooleanReturn" />
+		<module name="StringLiteralEquality" />
+		<!--<module name="NestedForDepth">-->
+			<!--<property name="max" value="3" />-->
+		<!--</module>-->
+		<!--<module name="NestedIfDepth">-->
+			<!--<property name="max" value="3" />-->
+		<!--</module>-->
+		<!--<module name="NestedTryDepth">-->
+			<!--<property name="max" value="3" />-->
+		<!--</module>-->
+		<module name="MultipleVariableDeclarations" />
+		<module name="RequireThis">
+			<property name="checkMethods" value="false" />
+		</module>
+		<module name="OneStatementPerLine" />
+
+		<!-- Imports -->
+		<module name="AvoidStarImport" />
+		<module name="AvoidStaticImport">
+			<property name="excludes"
+				value="org.assertj.core.api.Assertions.*, org.junit.Assert.*, org.junit.Assume.*, org.junit.internal.matchers.ThrowableMessageMatcher.*, org.hamcrest.CoreMatchers.*, org.hamcrest.Matchers.*, org.springframework.boot.configurationprocessor.ConfigurationMetadataMatchers.*, org.springframework.boot.configurationprocessor.TestCompiler.*, org.mockito.Mockito.*, org.mockito.BDDMockito.*, org.mockito.Matchers.*, org.springframework.test.web.servlet.request.MockMvcRequestBuilders.*, org.springframework.test.web.servlet.result.MockMvcResultMatchers.*, org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.*, org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.*, org.springframework.hateoas.mvc.ControllerLinkBuilder.linkTo" />
+		</module>
+		<module name="IllegalImport" />
+		<module name="RedundantImport" />
+		<module name="UnusedImports">
+			<property name="processJavadoc" value="true" />
+		</module>
+		<module name="ImportOrder">
+			<property name="groups" value="java,/^javax?\./,org,org.springframework,*" />
+			<property name="ordered" value="true" />
+			<property name="separated" value="true" />
+			<property name="option" value="top" />
+			<property name="sortStaticImportsAlphabetically" value="true" />
+		</module>
+
+		<!-- Javadoc Comments -->
+<!--		<module name="JavadocType"> -->
+<!--			<property name="scope" value="package"/> -->
+<!--			<property name="authorFormat" value=".+\s.+"/> -->
+<!--		</module> -->
+<!--		<module name="JavadocMethod"> -->
+<!--			<property name="allowMissingJavadoc" value="true" /> -->
+<!--		</module> -->
+<!--		<module name="JavadocVariable"> -->
+<!--			<property name="scope" value="public"/> -->
+<!--		</module> -->
+<!--		<module name="JavadocStyle"> -->
+<!--			<property name="checkEmptyJavadoc" value="true"/> -->
+<!--		</module> -->
+<!--		<module name="NonEmptyAtclauseDescription" /> -->
+<!--		<module name="JavadocTagContinuationIndentation"> -->
+<!--			<property name="offset" value="0"/> -->
+<!--		</module> -->
+<!--		<module name="AtclauseOrder"> -->
+<!--			<property name="target" value="CLASS_DEF, INTERFACE_DEF, ENUM_DEF"/> -->
+<!--		<property name="tagOrder" value="@param, @author, @since, @see, @version, @serial, @deprecated"/> -->
+<!--		</module> -->
+<!--		<module name="AtclauseOrder"> -->
+<!--			<property name="target" value="METHOD_DEF, CTOR_DEF, VARIABLE_DEF"/> -->
+<!--		<property name="tagOrder" value="@param, @return, @throws, @since, @deprecated, @see"/> -->
+<!--		</module> -->
+
+		<!-- Miscellaneous -->
+<!--		<module name="CommentsIndentation" /> -->
+<!--		<module name="UpperEll" /> -->
+<!--		<module name="ArrayTypeStyle" /> -->
+<!--		<module name="OuterTypeFilename" /> -->
+
+		<!-- Modifiers -->
+<!--		<module name="RedundantModifier" /> -->
+
+		<!-- Regexp -->
+<!--		<module name="RegexpSinglelineJava"> -->
+<!--			<property name="format" value="^\t* +\t*\S" /> -->
+<!--			<property name="message" -->
+<!--				value="Line has leading space characters; indentation should be performed with tabs only." /> -->
+<!--			<property name="ignoreComments" value="true" /> -->
+<!--		</module> -->
+<!--		<module name="RegexpSinglelineJava"> -->
+<!--			<property name="maximum" value="0"/> -->
+<!--			<property name="format" value="org\.mockito\.Mockito\.(when|doThrow|doAnswer)" /> -->
+<!--			<property name="message" -->
+<!--				value="Please use BDDMockto imports." /> -->
+<!--			<property name="ignoreComments" value="true" /> -->
+<!--		</module> -->
+<!--		<module name="RegexpSinglelineJava"> -->
+<!--			<property name="maximum" value="0"/> -->
+<!--			<property name="format" value="org\.junit\.Assert\.assert" /> -->
+<!--			<property name="message" -->
+<!--				value="Please use AssertJ imports." /> -->
+<!--			<property name="ignoreComments" value="true" /> -->
+<!--		</module> -->
+<!--		<module name="Regexp"> -->
+<!--			<property name="format" value="[ \t]+$" /> -->
+<!--			<property name="illegalPattern" value="true" /> -->
+<!--			<property name="message" value="Trailing whitespace" /> -->
+<!--		</module> -->
+
+		<!-- Whitespace -->
+		<module name="GenericWhitespace" />
+		<module name="MethodParamPad" />
+		<module name="NoWhitespaceAfter" >
+			<property name="tokens" value="BNOT, DEC, DOT, INC, LNOT, UNARY_MINUS, UNARY_PLUS, ARRAY_DECLARATOR"/>
+		</module>
+		<module name="NoWhitespaceBefore" />
+		<module name="ParenPad" />
+		<module name="TypecastParenPad" />
+		<module name="WhitespaceAfter" />
+		<module name="WhitespaceAround" />
+
+	</module>
+</module>

src/reference/asciidoc/quick-tour.adoc

@@ -72,7 +72,8 @@ public void testAutoCommit() throws Exception {
 private KafkaMessageListenerContainer<Integer, String> createContainer() {
     Map<String, Object> props = consumerProps();
     DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
-    KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf, topic1);
+    KafkaMessageListenerContainer<Integer, String> container =
+                                      new KafkaMessageListenerContainer<>(cf, topic1);
     return container;
 }
 
@@ -137,7 +138,8 @@ public class Config {
     @Bean
     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
                         kafkaListenerContainerFactory() {
-        SimpleKafkaListenerContainerFactory<Integer, String> factory = new SimpleKafkaListenerContainerFactory<>();
+        SimpleKafkaListenerContainerFactory<Integer, String> factory =
+                            new SimpleKafkaListenerContainerFactory<>();
         factory.setConsumerFactory(consumerFactory());
         return factory;
     }
@@ -151,13 +153,7 @@ public class Config {
     public Map<String, Object> consumerConfigs() {
         Map<String, Object> props = new HashMap<>();
         props.put("bootstrap.servers", embeddedKafka.getBrokersAsString());
-        props.put("bootstrap.servers", "localhost:9092");
-        props.put("group.id", "myGroup");
-        props.put("enable.auto.commit", true);
-        props.put("auto.commit.interval.ms", "100");
-        props.put("session.timeout.ms", "15000");
-        props.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
-        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+        ......
         return props;
     }
 
@@ -175,13 +171,7 @@ public class Config {
     public Map<String, Object> producerConfigs() {
         Map<String, Object> props = new HashMap<>();
         props.put("bootstrap.servers", embeddedKafka.getBrokersAsString());
-        props.put("bootstrap.servers", "localhost:9092");
-        props.put("retries", 0);
-        props.put("batch.size", 16384);
-        props.put("linger.ms", 1);
-        props.put("buffer.memory", 33554432);
-        props.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
-        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+        ......
         return props;
     }
 
@@ -196,7 +186,7 @@ public class Listener {
 
     private final CountDownLatch latch1 = new CountDownLatch(1);
 
-    @KafkaListener(id="foo", topics = "annotated1")
+    @KafkaListener(id = "foo", topics = "annotated1")
     public void listen1(String foo) {
         this.latch1.countDown();
     }