title: Testing Spring Cloud Stream app with kafka-streams binder
body: <p>After trying a number of different setups, and reading many different guides, I'm still at a loss for how to write simple unit tests for Spring Cloud Stream processors that use the kafka-streams binder.</p>
  <p>Starting with a completely new repo and following the setup outlined <a href="https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.6.RELEASE/reference/html/spring-cloud-stream.html#spring-cloud-stream-overview-introducing" rel="nofollow noreferrer">here</a>, I am able to write the basic uppercase demo function (in my code named 'uppercaseSimple') like so:</p>
  <pre><code>@Bean
  public Function&lt;String, String&gt; uppercaseSimple() {
      return value -&gt; value.toUpperCase();
  }
  </code></pre>
  <p>with a functioning test that passes:</p>
  <pre><code>@SpringBootTest(classes = SpringCloudStreamDemoApplication.class)
  @Import({TestChannelBinderConfiguration.class})
  class SpringCloudStreamDemoApplicationTests {
  
      @Autowired
      private InputDestination input;
  
      @Autowired
      private OutputDestination output;
  
      @Test
      void contextLoads() {
          input.send(new GenericMessage&lt;byte[]&gt;(&quot;hello&quot;.getBytes()));
          assertThat(output.receive().getPayload()).isEqualTo(&quot;HELLO&quot;.getBytes());
      }
  
  }
  </code></pre>
  <p>However, trying to change the function signature to a processor that makes full use of the kafka-streams binder like so:</p>
  <pre><code>@Bean
  public Function&lt;KStream&lt;Bytes, String&gt;, KStream&lt;Bytes, String&gt;&gt; uppercaseStream() {
      return input -&gt; input
          .map((key, value) -&gt; new KeyValue&lt;&gt;(null, value.toUpperCase()));
  }
  </code></pre>
  <p>means the same test now fails with the following error:</p>
  <pre><code>Index: 0, Size: 0
  java.lang.IndexOutOfBoundsException: Index: 0, Size: 0
      at java.util.ArrayList.rangeCheck(ArrayList.java:657)
      at java.util.ArrayList.get(ArrayList.java:433)
      at org.springframework.cloud.stream.binder.test.AbstractDestination.getChannel(AbstractDestination.java:34)
      at org.springframework.cloud.stream.binder.test.InputDestination.send(InputDestination.java:37)
      at com.example.springcloudstreamdemo.SpringCloudStreamDemoApplicationTests.contextLoads(SpringCloudStreamDemoApplicationTests.java:26)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:498)
      at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:686)
      at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
      at org.junit.jupiter.engine.execution.InvocationInterceptorChain$ValidatingInvocation.proceed(InvocationInterceptorChain.java:131)
      at org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:149)
      at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestableMethod(TimeoutExtension.java:140)
      at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestMethod(TimeoutExtension.java:84)
      at org.junit.jupiter.engine.execution.ExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(ExecutableInvoker.java:115)
      at org.junit.jupiter.engine.execution.ExecutableInvoker.lambda$invoke$0(ExecutableInvoker.java:105)
      at org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
      at org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
      at org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
      at org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
      at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:104)
      at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:98)
      at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeTestMethod$6(TestMethodTestDescriptor.java:212)
      at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
      at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.invokeTestMethod(TestMethodTestDescriptor.java:208)
      at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:137)
      at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:71)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$5(NodeTestTask.java:135)
      at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$7(NodeTestTask.java:125)
      at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:135)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:123)
      at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:122)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:80)
      at java.util.ArrayList.forEach(ArrayList.java:1257)
      at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:38)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$5(NodeTestTask.java:139)
      at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$7(NodeTestTask.java:125)
      at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:135)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:123)
      at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:122)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:80)
      at java.util.ArrayList.forEach(ArrayList.java:1257)
      at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:38)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$5(NodeTestTask.java:139)
      at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$7(NodeTestTask.java:125)
      at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:135)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:123)
      at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:122)
      at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:80)
      at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.submit(SameThreadHierarchicalTestExecutorService.java:32)
      at org.junit.platform.engine.support.hierarchical.HierarchicalTestExecutor.execute(HierarchicalTestExecutor.java:57)
      at org.junit.platform.engine.support.hierarchical.HierarchicalTestEngine.execute(HierarchicalTestEngine.java:51)
      at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:248)
      at org.junit.platform.launcher.core.DefaultLauncher.lambda$execute$5(DefaultLauncher.java:211)
      at org.junit.platform.launcher.core.DefaultLauncher.withInterceptedStreams(DefaultLauncher.java:226)
      at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:199)
      at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:132)
      at org.gradle.api.internal.tasks.testing.junitplatform.JUnitPlatformTestClassProcessor$CollectAllTestClassesExecutor.processAllTestClasses(JUnitPlatformTestClassProcessor.java:99)
      at org.gradle.api.internal.tasks.testing.junitplatform.JUnitPlatformTestClassProcessor$CollectAllTestClassesExecutor.access$000(JUnitPlatformTestClassProcessor.java:79)
      at org.gradle.api.internal.tasks.testing.junitplatform.JUnitPlatformTestClassProcessor.stop(JUnitPlatformTestClassProcessor.java:75)
      at org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.stop(SuiteTestClassProcessor.java:61)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:498)
      at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36)
      at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
      at org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:33)
      at org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:94)
      at com.sun.proxy.$Proxy2.stop(Unknown Source)
      at org.gradle.api.internal.tasks.testing.worker.TestWorker.stop(TestWorker.java:133)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:498)
      at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36)
      at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
      at org.gradle.internal.remote.internal.hub.MessageHubBackedObjectConnection$DispatchWrapper.dispatch(MessageHubBackedObjectConnection.java:182)
      at org.gradle.internal.remote.internal.hub.MessageHubBackedObjectConnection$DispatchWrapper.dispatch(MessageHubBackedObjectConnection.java:164)
      at org.gradle.internal.remote.internal.hub.MessageHub$Handler.run(MessageHub.java:414)
      at org.gradle.internal.concurrent.ExecutorPolicy$CatchAndRecordFailures.onExecute(ExecutorPolicy.java:64)
      at org.gradle.internal.concurrent.ManagedExecutorImpl$1.run(ManagedExecutorImpl.java:48)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at org.gradle.internal.concurrent.ThreadFactoryImpl$ManagedThreadRunnable.run(ThreadFactoryImpl.java:56)
      at java.lang.Thread.run(Thread.java:748)
  </code></pre>
  <p>I have pushed this demo to a GitHub <a href="https://github.com/terryf82/spring-cloud-stream-demo" rel="nofollow noreferrer">repo</a>, and have been toggling between testing the two functions by commenting in / out lines <a href="https://github.com/terryf82/spring-cloud-stream-demo/blob/master/src/main/java/com/example/springcloudstreamdemo/SpringCloudStreamDemoApplication.java#L19" rel="nofollow noreferrer">19</a> and <a href="https://github.com/terryf82/spring-cloud-stream-demo/blob/master/src/main/java/com/example/springcloudstreamdemo/SpringCloudStreamDemoApplication.java#L24" rel="nofollow noreferrer">24</a>, to ensure that only one version of the function is registered at a time. Where am I going wrong here? Is this kind of testing not applicable to functions that make use of KStream / KTable? The docs on testing seem to make no distinction about which underlying binder(s) are compatible, but it seems like kafka-streams isn't?</p>