GH-598: ConsumerFactory default methods

Avoid implementations having to implement all methods.

Resolve a couple TODOs

Add default implementation for `ConsumerFactory.createConsumer()` methods

Fixes #598

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -28,14 +28,17 @@
  * @param <V> the value type.
  *
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public interface ConsumerFactory<K, V> {
 
 	/**
 	 * Create a consumer with the group id and client id as configured in the properties.
 	 * @return the consumer.
 	 */
-	Consumer<K, V> createConsumer();
+	default Consumer<K, V> createConsumer() {
+		return createConsumer(null);
+	}
 
 	/**
 	 * Create a consumer, appending the suffix to the {@code client.id} property,
@@ -44,7 +47,9 @@ public interface ConsumerFactory<K, V> {
 	 * @return the consumer.
 	 * @since 1.3
 	 */
-	Consumer<K, V> createConsumer(String clientIdSuffix);
+	default Consumer<K, V> createConsumer(String clientIdSuffix) {
+		return createConsumer(null, clientIdSuffix);
+	}
 
 	/**
 	 * Create a consumer with an explicit group id; in addition, the
@@ -55,24 +60,23 @@ public interface ConsumerFactory<K, V> {
 	 * @return the consumer.
 	 * @since 1.3
 	 */
-	Consumer<K, V> createConsumer(String groupId, String clientIdSuffix);
+	default Consumer<K, V> createConsumer(String groupId, String clientIdSuffix) {
+		return createConsumer(groupId, null, clientIdSuffix);
+	}
 
 	/**
 	 * Create a consumer with an explicit group id; in addition, the
 	 * client id suffix is appended to the clientIdPrefix which overrides the
 	 * {@code client.id} property, if present.
 	 * If a factory does not implement this method, {@link #createConsumer(String, String)}
 	 * is invoked, ignoring the prefix.
-	 * TODO: remove default in 2.2
 	 * @param groupId the group id.
 	 * @param clientIdPrefix the prefix.
 	 * @param clientIdSuffix the suffix.
 	 * @return the consumer.
 	 * @since 2.1.1
 	 */
-	default Consumer<K, V> createConsumer(String groupId, String clientIdPrefix, String clientIdSuffix) {
-		return createConsumer(groupId, clientIdSuffix);
-	}
+	Consumer<K, V> createConsumer(String groupId, String clientIdPrefix, String clientIdSuffix);
 
 	/**
 	 * Return true if consumers created by this factory use auto commit.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -38,6 +38,7 @@
  *
  * @author Gary Russell
  * @author Murali Reddy
+ * @author Artem Bilan
  */
 public class DefaultKafkaConsumerFactory<K, V> implements ConsumerFactory<K, V> {
 
@@ -82,44 +83,22 @@ public Deserializer<V> getValueDeserializer() {
 		return this.valueDeserializer;
 	}
 
-	@Override
-	public Consumer<K, V> createConsumer() {
-		return createKafkaConsumer();
-	}
-
-	@Override
-	public Consumer<K, V> createConsumer(String clientIdSuffix) {
-		return createKafkaConsumer(null, clientIdSuffix);
-	}
-
-	@Override
-	public Consumer<K, V> createConsumer(String groupId, String clientIdSuffix) {
-		return createKafkaConsumer(groupId, clientIdSuffix);
-	}
-
 	@Override
 	public Consumer<K, V> createConsumer(String groupId, String clientIdPrefix, String clientIdSuffix) {
 		return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffix);
 	}
 
-	protected KafkaConsumer<K, V> createKafkaConsumer() {
-		return createKafkaConsumer(this.configs);
-	}
-
-	protected KafkaConsumer<K, V> createKafkaConsumer(String groupId, String clientIdSuffix) {
-		return createKafkaConsumer(groupId, null, clientIdSuffix);
-	}
-
 	protected KafkaConsumer<K, V> createKafkaConsumer(String groupId, String clientIdPrefix,
 			String clientIdSuffix) {
+
 		boolean overrideClientIdPrefix = StringUtils.hasText(clientIdPrefix);
 		if (clientIdSuffix == null) {
 			clientIdSuffix = "";
 		}
 		boolean shouldModifyClientId = (this.configs.containsKey(ConsumerConfig.CLIENT_ID_CONFIG)
 				&& StringUtils.hasText(clientIdSuffix)) || overrideClientIdPrefix;
 		if (groupId == null && !shouldModifyClientId) {
-			return createKafkaConsumer();
+			return createKafkaConsumer(this.configs);
 		}
 		else {
 			Map<String, Object> modifiedConfigs = new HashMap<>(this.configs);
@@ -136,7 +115,7 @@ protected KafkaConsumer<K, V> createKafkaConsumer(String groupId, String clientI
 	}
 
 	protected KafkaConsumer<K, V> createKafkaConsumer(Map<String, Object> configs) {
-		return new KafkaConsumer<K, V>(configs, this.keyDeserializer, this.valueDeserializer);
+		return new KafkaConsumer<>(configs, this.keyDeserializer, this.valueDeserializer);
 	}
 
 	@Override

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

@@ -34,6 +34,7 @@
 import org.springframework.kafka.support.KafkaHeaderMapper;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.KafkaNull;
+import org.springframework.kafka.support.SimpleKafkaHeaderMapper;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.MessageBuilder;
@@ -65,7 +66,9 @@ public MessagingMessageConverter() {
 		if (JacksonPresent.isJackson2Present()) {
 			this.headerMapper = new DefaultKafkaHeaderMapper();
 		}
-		// TODO: In 2.2, default to SimpleKafkaHeaderMapper if no Jackson
+		else {
+			this.headerMapper = new SimpleKafkaHeaderMapper();
+		}
 	}
 
 	/**