GH-1787: Only Validate Once

Resolves #1787

The validator was invoked twice with a class-level `@KafkaListener` when
no default `@KafkaHandler` is present.

Author: garyrussell

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

@@ -204,7 +204,7 @@ public Object getBean() {
 	public Object invoke(Message<?> message, Object... providedArgs) throws Exception { //NOSONAR
 		Class<? extends Object> payloadClass = message.getPayload().getClass();
 		InvocableHandlerMethod handler = getHandlerForPayload(payloadClass);
-		if (this.validator != null) {
+		if (this.validator != null && this.defaultHandler != null) {
 			MethodParameter parameter = this.payloadMethodParameters.get(handler);
 			if (parameter != null) {
 				this.validator.validate(message, parameter, message.getPayload());

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

@@ -177,7 +177,7 @@
 		"annotated25", "annotated25reply1", "annotated25reply2", "annotated26", "annotated27", "annotated28",
 		"annotated29", "annotated30", "annotated30reply", "annotated31", "annotated32", "annotated33",
 		"annotated34", "annotated35", "annotated36", "annotated37", "foo", "manualStart", "seekOnIdle",
-		"annotated38", "annotated38reply", "annotated39" })
+		"annotated38", "annotated38reply", "annotated39", "annotated40" })
 public class EnableKafkaIntegrationTests {
 
 	private static final String DEFAULT_TEST_GROUP_ID = "testAnnot";
@@ -202,6 +202,9 @@ public class EnableKafkaIntegrationTests {
 	@Autowired
 	public MultiJsonListenerBean multiJsonListener;
 
+	@Autowired
+	public MultiListenerNoDefault multiNoDefault;
+
 	@Autowired
 	public KafkaTemplate<Integer, String> template;
 
@@ -446,7 +449,19 @@ public void testMultiJson() throws Exception {
 		assertThat(this.multiJsonListener.baz.getBar()).isEqualTo("two");
 		assertThat(this.multiJsonListener.bar.getBar()).isEqualTo("three");
 		assertThat(this.multiJsonListener.bar).isInstanceOf(Qux.class);
+		assertThat(this.multiJsonListener.validated).isNotNull();
 		assertThat(this.multiJsonListener.validated.isValidated()).isTrue();
+		assertThat(this.multiJsonListener.validated.valCount).isEqualTo(1);
+	}
+
+	@Test
+	public void testMultiValidateNoDefaultHandler() throws Exception {
+		this.kafkaJsonTemplate.setDefaultTopic("annotated40");
+		this.kafkaJsonTemplate.send(new GenericMessage<>(new ValidatedClass(5)));
+		assertThat(this.multiNoDefault.latch1.await(60, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.multiNoDefault.validated).isNotNull();
+		assertThat(this.multiNoDefault.validated.isValidated()).isTrue();
+		assertThat(this.multiNoDefault.validated.valCount).isEqualTo(1);
 	}
 
 	@Test
@@ -1306,6 +1321,11 @@ public MultiJsonListenerBean multiJsonListener() {
 			return new MultiJsonListenerBean();
 		}
 
+		@Bean
+		public MultiListenerNoDefault multiNoDefault() {
+			return new MultiListenerNoDefault();
+		}
+
 		@Bean
 		public MultiListenerSendTo multiListenerSendTo() {
 			return new MultiListenerSendTo();
@@ -2218,6 +2238,21 @@ public void defaultHandler(Bar bar) {
 
 	}
 
+	@KafkaListener(id = "multiNoDefault", topics = "annotated40", containerFactory = "kafkaJsonListenerContainerFactory2")
+	static class MultiListenerNoDefault {
+
+		final CountDownLatch latch1 = new CountDownLatch(1);
+
+		volatile ValidatedClass validated;
+
+		@KafkaHandler
+		public void bar(@Valid ValidatedClass val) {
+			this.validated = val;
+			this.latch1.countDown();
+		}
+
+	}
+
 	@KafkaListener(id = "multiSendTo", topics = "annotated25")
 	@SendTo("annotated25reply1")
 	static class MultiListenerSendTo {
@@ -2348,6 +2383,7 @@ public static class ValidatedClass {
 
 		private volatile boolean validated;
 
+		volatile int valCount;
 
 		public ValidatedClass() {
 		}
@@ -2370,6 +2406,9 @@ public boolean isValidated() {
 
 		public void setValidated(boolean validated) {
 			this.validated = validated;
+			if (validated) { // don't count the json deserialization call
+				this.valCount++;
+			}
 		}
 
 	}