184184 "annotated25" , "annotated25reply1" , "annotated25reply2" , "annotated26" , "annotated27" , "annotated28" ,
185185 "annotated29" , "annotated30" , "annotated30reply" , "annotated31" , "annotated32" , "annotated33" ,
186186 "annotated34" , "annotated35" , "annotated36" , "annotated37" , "foo" , "manualStart" , "seekOnIdle" ,
187- "annotated38" , "annotated38reply" , "annotated39" , "annotated40" , "annotated41" , "annotated42" })
187+ "annotated38" , "annotated38reply" , "annotated39" , "annotated40" , "annotated41" , "annotated42" ,
188+ "annotated43" , "annotated43reply" })
188189@ TestPropertySource (properties = "spel.props=fetch.min.bytes=420000,max.poll.records=10" )
189190public class EnableKafkaIntegrationTests {
190191
@@ -438,6 +439,15 @@ public void testInterface() throws Exception {
438439 template .send ("annotated7" , 0 , "foo" );
439440 template .flush ();
440441 assertThat (this .ifaceListener .getLatch1 ().await (60 , TimeUnit .SECONDS )).isTrue ();
442+ Map <String , Object > consumerProps = new HashMap <>(this .consumerFactory .getConfigurationProperties ());
443+ consumerProps .put (ConsumerConfig .GROUP_ID_CONFIG , "testInterface" );
444+ ConsumerFactory <Integer , String > cf = new DefaultKafkaConsumerFactory <>(consumerProps );
445+ Consumer <Integer , String > consumer = cf .createConsumer ();
446+ this .embeddedKafka .consumeFromAnEmbeddedTopic (consumer , "annotated43reply" );
447+ template .send ("annotated43" , 0 , "foo" );
448+ ConsumerRecord <Integer , String > reply = KafkaTestUtils .getSingleRecord (consumer , "annotated43reply" );
449+ assertThat (reply ).extracting (rec -> rec .value ()).isEqualTo ("FOO" );
450+ consumer .close ();
441451 }
442452
443453 @ Test
@@ -1375,8 +1385,8 @@ public MultiListenerNoDefault multiNoDefault() {
13751385 }
13761386
13771387 @ Bean
1378- public MultiListenerSendTo multiListenerSendTo () {
1379- return new MultiListenerSendTo ();
1388+ public MultiListenerSendToImpl multiListenerSendTo () {
1389+ return new MultiListenerSendToImpl ();
13801390 }
13811391
13821392 @ Bean
@@ -2290,6 +2300,10 @@ interface IfaceListener<T> {
22902300
22912301 void listen (T foo );
22922302
2303+ @ SendTo ("annotated43reply" )
2304+ @ KafkaListener (id = "ifcR" , topics = "annotated43" )
2305+ String reply (String in );
2306+
22932307 }
22942308
22952309 static class IfaceListenerImpl implements IfaceListener <String > {
@@ -2310,6 +2324,11 @@ public void listenTx(String foo) {
23102324 latch2 .countDown ();
23112325 }
23122326
2327+ @ Override
2328+ public String reply (String in ) {
2329+ return in .toUpperCase ();
2330+ }
2331+
23132332 public CountDownLatch getLatch1 () {
23142333 return latch1 ;
23152334 }
@@ -2416,15 +2435,25 @@ public void bar(@Valid ValidatedClass val) {
24162435
24172436 @ KafkaListener (id = "multiSendTo" , topics = "annotated25" )
24182437 @ SendTo ("annotated25reply1" )
2419- static class MultiListenerSendTo {
2438+ interface MultiListenerSendTo {
2439+
2440+ @ KafkaHandler
2441+ String foo (String in );
24202442
24212443 @ KafkaHandler
2444+ @ SendTo ("!{'annotated25reply2'}" )
2445+ String bar (KafkaNull nul , int key );
2446+
2447+ }
2448+
2449+ static class MultiListenerSendToImpl implements MultiListenerSendTo {
2450+
2451+ @ Override
24222452 public String foo (String in ) {
24232453 return in .toUpperCase ();
24242454 }
24252455
2426- @ KafkaHandler
2427- @ SendTo ("!{'annotated25reply2'}" )
2456+ @ Override
24282457 public String bar (@ Payload (required = false ) KafkaNull nul ,
24292458 @ Header (KafkaHeaders .RECEIVED_MESSAGE_KEY ) int key ) {
24302459 return "BAR" ;
0 commit comments