GH-335: Fix Class Cast with Batch Filter

Fixes #335

Failed to check `batchListener` field when casting listener to apply filter.

Conflicts:
	spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java
	spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
Resolved.

Author: garyrussell

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

@@ -30,9 +30,13 @@
 import org.springframework.beans.factory.config.BeanExpressionResolver;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.kafka.listener.AcknowledgingMessageListener;
+import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
+import org.springframework.kafka.listener.BatchMessageListener;
 import org.springframework.kafka.listener.MessageListener;
 import org.springframework.kafka.listener.MessageListenerContainer;
 import org.springframework.kafka.listener.adapter.FilteringAcknowledgingMessageListenerAdapter;
+import org.springframework.kafka.listener.adapter.FilteringBatchAcknowledgingMessageListenerAdapter;
+import org.springframework.kafka.listener.adapter.FilteringBatchMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
@@ -313,10 +317,19 @@ private void setupMessageListener(MessageListenerContainer container, MessageCon
 						(AcknowledgingMessageListener<K, V>) messageListener, this.recordFilterStrategy,
 						this.ackDiscarded);
 			}
-			else {
+			else if (messageListener instanceof MessageListener) {
 				messageListener = new FilteringMessageListenerAdapter<>((MessageListener<K, V>) messageListener,
 						this.recordFilterStrategy);
 			}
+			else if (messageListener instanceof BatchAcknowledgingMessageListener) {
+				messageListener = new FilteringBatchAcknowledgingMessageListenerAdapter<>(
+						(BatchAcknowledgingMessageListener<K, V>) messageListener, this.recordFilterStrategy,
+						this.ackDiscarded);
+			}
+			else if (messageListener instanceof BatchMessageListener) {
+				messageListener = new FilteringBatchMessageListenerAdapter<>(
+						(BatchMessageListener<K, V>) messageListener, this.recordFilterStrategy);
+			}
 		}
 		container.setupMessageListener(messageListener);
 	}

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

@@ -120,10 +120,11 @@ public class EnableKafkaIntegrationTests {
 	public KafkaListenerEndpointRegistry registry;
 
 	@Autowired
-	private RecordFilterImpl recordFilter;
+	private RecordPassAllFilter recordFilter;
 
 	@Test
 	public void testSimple() throws Exception {
+		this.recordFilter.called = false;
 		template.send("annotated1", 0, "foo");
 		template.flush();
 		assertThat(this.listener.latch1.await(60, TimeUnit.SECONDS)).isTrue();
@@ -251,6 +252,7 @@ public void testEmpty() throws Exception {
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testBatch() throws Exception {
+		this.recordFilter.called = false;
 		ConcurrentMessageListenerContainer<?, ?> container =
 				(ConcurrentMessageListenerContainer<?, ?>) registry.getListenerContainer("list1");
 		Consumer<?, ?> consumer =
@@ -277,6 +279,7 @@ public void testBatch() throws Exception {
 		List<?> list = (List<?>) this.listener.payload;
 		assertThat(list.size()).isGreaterThan(0);
 		assertThat(list.get(0)).isInstanceOf(String.class);
+		assertThat(this.recordFilter.called).isTrue();
 
 		assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
 	}
@@ -388,13 +391,13 @@ public PlatformTransactionManager transactionManager() {
 		}
 
 		@Bean
-		public RecordFilterImpl recordFilter() {
-			return new RecordFilterImpl();
+		public RecordPassAllFilter recordFilter() {
+			return new RecordPassAllFilter();
 		}
 
 		@Bean
-		public RecordFilterImpl manualFilter() {
-			return new RecordFilterImpl();
+		public RecordPassAllFilter manualFilter() {
+			return new RecordPassAllFilter();
 		}
 
 		@Bean
@@ -412,6 +415,7 @@ public KafkaListenerContainerFactory<?> batchFactory() {
 					new ConcurrentKafkaListenerContainerFactory<>();
 			factory.setConsumerFactory(consumerFactory());
 			factory.setBatchListener(true);
+			factory.setRecordFilterStrategy(recordFilter());
 			return factory;
 		}
 
@@ -821,7 +825,7 @@ public void setBar(String bar) {
 
 	}
 
-	public static class RecordFilterImpl implements RecordFilterStrategy<Integer, String> {
+	public static class RecordPassAllFilter implements RecordFilterStrategy<Integer, String> {
 
 		private boolean called;