title: Using Spring Cloud Stream to produce Avro Messages using the Confluent Schema Registry
body: <p>I am new to Spring Cloud Stream and trying to produce Avro messages based on the Confluent Schema Registry.</p>
  
  <p>I can get the very basic example, <a href="https://cloud.spring.io/spring-cloud-stream/" rel="nofollow noreferrer">https://cloud.spring.io/spring-cloud-stream/</a> working, but when I try to extend it further to use Avro, I am getting exceptions.</p>
  
  <p>As a generic test environment, I am running the Docker image:
  <a href="https://github.com/Landoop/fast-data-dev/" rel="nofollow noreferrer">https://github.com/Landoop/fast-data-dev/</a></p>
  
  <p>The test project was generated and slightly modified using Spring Initializr:
  <a href="https://start.spring.io/" rel="nofollow noreferrer">https://start.spring.io/</a></p>
  
  <p><strong>The pom.xml:</strong></p>
  
  <pre><code>&lt;?xml version="1.0" encoding="UTF-8"?&gt;
  &lt;project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"&gt;
    &lt;modelVersion&gt;4.0.0&lt;/modelVersion&gt;
  
    &lt;groupId&gt;com.demo.kafka&lt;/groupId&gt;
    &lt;artifactId&gt;kafka-demo&lt;/artifactId&gt;
    &lt;version&gt;0.0.1-SNAPSHOT&lt;/version&gt;
    &lt;packaging&gt;jar&lt;/packaging&gt;
  
    &lt;name&gt;kafka-demo&lt;/name&gt;
    &lt;description&gt;Demo project for Spring Boot&lt;/description&gt;
  
  
    &lt;parent&gt;
      &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
      &lt;artifactId&gt;spring-boot-starter-parent&lt;/artifactId&gt;
      &lt;version&gt;1.5.4.RELEASE&lt;/version&gt;
      &lt;relativePath/&gt; &lt;!-- lookup parent from repository --&gt;
    &lt;/parent&gt;
  
    &lt;properties&gt;
      &lt;project.build.sourceEncoding&gt;UTF-8&lt;/project.build.sourceEncoding&gt;
      &lt;project.reporting.outputEncoding&gt;UTF-8&lt;/project.reporting.outputEncoding&gt;
      &lt;java.version&gt;1.8&lt;/java.version&gt;
      &lt;spring-cloud.version&gt;Dalston.SR1&lt;/spring-cloud.version&gt;
      &lt;project-avro.version&gt;7.0.0-SNAPSHOT&lt;/project-avro.version&gt;
      &lt;apache.avro.version&gt;1.8.2&lt;/apache.avro.version&gt;
    &lt;/properties&gt;
  
    &lt;dependencyManagement&gt;
      &lt;dependencies&gt;
        &lt;dependency&gt;
          &lt;groupId&gt;org.springframework.cloud&lt;/groupId&gt;
          &lt;artifactId&gt;spring-cloud-dependencies&lt;/artifactId&gt;
          &lt;version&gt;${spring-cloud.version}&lt;/version&gt;
          &lt;type&gt;pom&lt;/type&gt;
          &lt;scope&gt;import&lt;/scope&gt;
        &lt;/dependency&gt;
      &lt;/dependencies&gt;
    &lt;/dependencyManagement&gt;
  
    &lt;dependencies&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.springframework.cloud&lt;/groupId&gt;
        &lt;artifactId&gt;spring-cloud-stream&lt;/artifactId&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.springframework.cloud&lt;/groupId&gt;
        &lt;artifactId&gt;spring-cloud-starter-stream-kafka&lt;/artifactId&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.springframework.cloud&lt;/groupId&gt;
        &lt;artifactId&gt;spring-cloud-stream-schema&lt;/artifactId&gt;
        &lt;version&gt;1.2.2.RELEASE&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
        &lt;artifactId&gt;spring-boot-starter-test&lt;/artifactId&gt;
        &lt;scope&gt;test&lt;/scope&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;joda-time&lt;/groupId&gt;
        &lt;artifactId&gt;joda-time&lt;/artifactId&gt;
        &lt;version&gt;2.9.9&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.apache.avro&lt;/groupId&gt;
        &lt;artifactId&gt;avro&lt;/artifactId&gt;
        &lt;version&gt;${apache.avro.version}&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;project.foundation&lt;/groupId&gt;
        &lt;artifactId&gt;maritz-avro&lt;/artifactId&gt;
        &lt;version&gt;${project-avro.version}&lt;/version&gt;
      &lt;/dependency&gt;
    &lt;/dependencies&gt;
  &lt;/project&gt;
  </code></pre>
  
  <p><strong>I have published the following schema to the registry:</strong></p>
  
  <pre><code>{
    "type": "record",
    "name": "RoutingSlipMsg",
    "fields": [
      {
        "name": "projectNumber",
        "type": "string"
      },
      {
        "name": "clientName",
        "type": "string"
      },
      {
        "name": "programTheme",
        "type": "string"
      },
      {
        "name": "databaseName",
        "type": [
          "null",
          "string"
        ]
      },
      {
        "name": "actionType",
        "type": [
          "null",
          "string"
        ]
      }
    ]
  }
  </code></pre>
  
  <p><strong>Original Test application.yml</strong></p>
  
  <pre><code>spring:
    cloud:
      stream:
        schema:
          avro:
            dynamicSchemaGenerationEnabled: true
        bindings:
          output:
            binder: kafka
            destination: routing_slip_dev
  #          contentType: application/avro
  #          contentType: application/*+avro
            contentType: application/com.project.RoutingSlipMsg.v1+avro
        schemaRegistryClient:
          endpoint: http://192.168.99.100:8081
        kafka:
          binder:
            defaultBrokerPort: 9092
            defaultZkPort: 2181
            zkNodes: 192.168.99.100
            brokers: 192.168.99.100
  </code></pre>
  
  <p><strong>Test application.yml edits:</strong></p>
  
  <pre><code>spring:
    cloud:
      stream:
        schema:
          avro:
            dynamicSchemaGenerationEnabled: true
        bindings:
          output:
            binder: kafka
            destination: routing_slip_dev
  #          contentType: application/*+avro
  #          contentType: application/avro
        schemaRegistryClient:
          endpoint: http://192.168.99.100:8081
        kafka:
          binder:
            defaultBrokerPort: 9092
            defaultZkPort: 2181
            zkNodes: 192.168.99.100
            brokers: 192.168.99.100
  </code></pre>
  
  <p><strong>Configuration class</strong></p>
  
  <pre><code>@Configuration
  public class DemoConfiguration {
  
  
    @Bean
    public ConfluentSchemaRegistryClient schemaRegistryClient() {
      ConfluentSchemaRegistryClient registry = new ConfluentSchemaRegistryClient();
      registry.setEndpoint("http://192.168.99.100:8081");
      return registry;
    }
  
  }
  </code></pre>
  
  <p>The auto topic creation seems to be working properly.  I can change the <em>destination:</em> value to something else each time and a new topic will be created.</p>
  
  <p><strong>Test</strong></p>
  
  <pre><code>package com.project.kafka.kafkademo;
  
  import com.project.avro.RoutingSlipMsg;
  import org.junit.Test;
  import org.junit.runner.RunWith;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.boot.test.context.SpringBootTest;
  import org.springframework.test.context.junit4.SpringRunner;
  
  @RunWith(SpringRunner.class)
  @SpringBootTest
  public class KafkaDemoApplicationTests {
  
    @Autowired
    private KafkaRoutingSlipSender kafkaRoutingSlipSender;
  
    @Test
    public void testSend() throws Exception {
  
      RoutingSlipMsg routingSlip = new RoutingSlipMsg();
      routingSlip.setActionType("FAKE_ACTION");
      routingSlip.setClientName("TEST_CLIENT");
      routingSlip.setDatabaseName("NONE");
      routingSlip.setProgramTheme("THEME");
      routingSlip.setProjectNumber("ABC123");
  
      kafkaRoutingSlipSender.send(routingSlip);
    }
  }
  </code></pre>
  
  <p>**BootApplication class*</p>
  
  <pre><code>package com.project.kafka.kafkademo;
  
  import org.springframework.boot.SpringApplication;
  import org.springframework.boot.autoconfigure.SpringBootApplication;
  
  @SpringBootApplication
  public class KafkaDemoApplication {
  
    public static void main(String[] args) {
      SpringApplication.run(KafkaDemoApplication.class, args);
    }
  }
  </code></pre>
  
  <p><strong>Original Sender class</strong></p>
  
  <pre><code>package com.project.kafka.kafkademo;
  
  import com.project.avro.RoutingSlipMsg;
  import org.springframework.cloud.stream.annotation.EnableBinding;
  import org.springframework.cloud.stream.messaging.Source;
  import org.springframework.cloud.stream.schema.client.EnableSchemaRegistryClient;
  import org.springframework.integration.annotation.InboundChannelAdapter;
  import org.springframework.integration.core.MessageSource;
  import org.springframework.messaging.support.GenericMessage;
  import org.springframework.stereotype.Component;
  
  @Component
  @EnableBinding(Source.class)
  @EnableSchemaRegistryClient
  public class KafkaRoutingSlipSender {
  
  
    @InboundChannelAdapter(value = Source.OUTPUT)
    public MessageSource&lt;RoutingSlipMsg&gt; send(RoutingSlipMsg data) {
      System.out.println("**** Hello World ****");
      return () -&gt; {
        return new GenericMessage&lt;&gt;(data);
      };
    }
  }
  </code></pre>
  
  <p><strong>Sender Class Edits:</strong></p>
  
  <pre><code>@Component
  @EnableBinding(Source.class)
  @EnableSchemaRegistryClient
  public class KafkaRoutingSlipSender {
  
    @Autowired
    private Source source;
  
    public void send(RoutingSlipMsg data) {
      System.out.println("**** Hello World ****");
      source.output().send(MessageBuilder.withPayload(data).build());
    }
  }
  </code></pre>
  
  <p><strong>Original Stacktrace / Exception due to inappropriate use of <code>@InboundChannelAdapter</code></strong></p>
  
  <pre><code>2017-07-06 12:20:48.645 ERROR 69011 --- [ask-scheduler-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessagingException: Failed to invoke method; nested exception is java.lang.IllegalArgumentException: wrong number of arguments
      at org.springframework.integration.endpoint.MethodInvokingMessageSource.doReceive(MethodInvokingMessageSource.java:119)
      at org.springframework.integration.endpoint.AbstractMessageSource.receive(AbstractMessageSource.java:134)
      at org.springframework.integration.endpoint.SourcePollingChannelAdapter.receiveMessage(SourcePollingChannelAdapter.java:224)
      at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:245)
      at org.springframework.integration.endpoint.AbstractPollingEndpoint.access$000(AbstractPollingEndpoint.java:58)
      at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:190)
      at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:186)
      at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller$1.run(AbstractPollingEndpoint.java:353)
      at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:55)
      at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
      at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:51)
      at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller.run(AbstractPollingEndpoint.java:344)
      at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
      at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:81)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
      at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
      at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
      at java.lang.Thread.run(Thread.java:745)
  Caused by: java.lang.IllegalArgumentException: wrong number of arguments
      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.springframework.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:216)
      at org.springframework.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:201)
      at org.springframework.integration.endpoint.MethodInvokingMessageSource.doReceive(MethodInvokingMessageSource.java:116)
      ... 20 more
  </code></pre>
  
  <p>I have reviewed several different posts on here, but nothing seems to be helping.
  What might I be missing?</p>
  
  <p><strong>Update 1:</strong>
  I have made updates to the <code>KafkaRoutingSlipSender</code> class and the <code>application.yml</code> as shown above.  With the <code>contentType</code> removed, these changes will publish a message like:</p>
  
  <pre><code>ÿcontentType?"application/x-java-object;type=com.maritz.avro.RoutingSlipMsg"FAKE_ACTIOÎTEST_CLIENÔNONÅTHEMÅABC12³
  </code></pre>
  
  <p>to the topic.</p>
  
  <p>However, when I set the <code>contentType</code> to <code>application/*+avro</code>, the <code>@Autowired Source</code> appears to be casuing a <code>NullPointerException</code>.</p>
  
  <p><strong>New Stacktrace / Exception:</strong></p>
  
  <pre><code>org.springframework.messaging.MessageDeliveryException: failed to send Message to channel 'output'; nested exception is java.lang.NullPointerException
  , failedMessage=GenericMessage [payload={"projectNumber": "ABC123", "clientName": "TEST_CLIENT", "programTheme": "THEME", "databaseName": "NONE", "actionType": "FAKE_ACTION"}, headers={id=9a9ebc45-9431-8de5-1cb0-40d191082599, timestamp=1499438830456}]
      at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:449)
      at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373)
      at com.maritz.kafka.kafkademo.KafkaRoutingSlipSender.send(KafkaRoutingSlipSender.java:22)
      at com.maritz.kafka.kafkademo.KafkaDemoApplicationTests.testSend(KafkaDemoApplicationTests.java:51)
      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.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
      at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
      at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
      at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
      at org.springframework.test.context.junit4.statements.RunBeforeTestMethodCallbacks.evaluate(RunBeforeTestMethodCallbacks.java:75)
      at org.springframework.test.context.junit4.statements.RunAfterTestMethodCallbacks.evaluate(RunAfterTestMethodCallbacks.java:86)
      at org.springframework.test.context.junit4.statements.SpringRepeat.evaluate(SpringRepeat.java:84)
      at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
      at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:252)
      at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:94)
      at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
      at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
      at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
      at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
      at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
      at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
      at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
      at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
      at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:191)
      at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
      at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
      at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:51)
      at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
      at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
  Caused by: java.lang.NullPointerException
      at org.springframework.cloud.stream.schema.client.DefaultSchemaRegistryClient.register(DefaultSchemaRegistryClient.java:57)
      at org.springframework.cloud.stream.schema.avro.AvroSchemaRegistryClientMessageConverter.resolveSchemaForWriting(AvroSchemaRegistryClientMessageConverter.java:242)
      at org.springframework.cloud.stream.schema.avro.AbstractAvroMessageConverter.convertToInternal(AbstractAvroMessageConverter.java:174)
      at org.springframework.messaging.converter.AbstractMessageConverter.toMessage(AbstractMessageConverter.java:193)
      at org.springframework.cloud.stream.binding.MessageConverterConfigurer$ContentTypeConvertingInterceptor.preSend(MessageConverterConfigurer.java:253)
      at org.springframework.integration.channel.AbstractMessageChannel$ChannelInterceptorList.preSend(AbstractMessageChannel.java:538)
      at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:415)
      ... 31 more
  </code></pre>
  
  <p><strong>Update 2 (07-10-2017):</strong>
  I have fixed the <code>null</code> Source issue, thanks to this information:</p>
  
  <p><a href="https://stackoverflow.com/questions/41003355/spring-boot-configuration-class-is-simply-ignored-and-not-loaded">Spring Boot: Configuration Class is simply ignored and not loaded</a></p>
  
  <p>The <code>ConfluentSchemaRegistryClient</code> is now being used.  I can step into the code when debugging, but it doesn't find the schema.</p>
  
  <p><strong>ConfluentSchemaRegistryClient.register(String subject, String format, String schema)</strong></p>
  
  <pre><code>ResponseEntity&lt;Map&gt; response = this.template.exchange(this.endpoint + path, HttpMethod.POST, request, Map.class, new Object[0]);
  </code></pre>
  
  <p><strong>New Stacktrace / Exception (07-10-2017):</strong></p>
  
  <pre><code>org.springframework.messaging.MessageDeliveryException: failed to send Message to channel 'output'; nested exception is org.springframework.web.client.HttpClientErrorException: 404 Not Found
  , failedMessage=GenericMessage [payload={"projectNumber": "ABC123", "clientName": "TEST_CLIENT", "programTheme": "THEME", "databaseName": "NONE", "actionType": "FAKE_ACTION"}, headers={id=73365130-a91b-ba9c-aaf0-77cfaa26f73d, timestamp=1499700409997}]
      at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:449)
      at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373)
      at com.maritz.kafka.kafkademo.KafkaRoutingSlipSender.send(KafkaRoutingSlipSender.java:29)
      at com.maritz.kafka.kafkademo.KafkaDemoApplicationTests.testSend(KafkaDemoApplicationTests.java:52)
      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.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
      at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
      at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
      at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
      at org.springframework.test.context.junit4.statements.RunBeforeTestMethodCallbacks.evaluate(RunBeforeTestMethodCallbacks.java:75)
      at org.springframework.test.context.junit4.statements.RunAfterTestMethodCallbacks.evaluate(RunAfterTestMethodCallbacks.java:86)
      at org.springframework.test.context.junit4.statements.SpringRepeat.evaluate(SpringRepeat.java:84)
      at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
      at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:252)
      at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:94)
      at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
      at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
      at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
      at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
      at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
      at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
      at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
      at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
      at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:191)
      at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
      at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
      at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:51)
      at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
      at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
  Caused by: org.springframework.web.client.HttpClientErrorException: 404 Not Found
      at org.springframework.web.client.DefaultResponseErrorHandler.handleError(DefaultResponseErrorHandler.java:63)
      at org.springframework.web.client.RestTemplate.handleResponse(RestTemplate.java:700)
      at org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:653)
      at org.springframework.web.client.RestTemplate.execute(RestTemplate.java:613)
      at org.springframework.web.client.RestTemplate.exchange(RestTemplate.java:531)
      at org.springframework.cloud.stream.schema.client.ConfluentSchemaRegistryClient.register(ConfluentSchemaRegistryClient.java:73)
      at org.springframework.cloud.stream.schema.avro.AvroSchemaRegistryClientMessageConverter.resolveSchemaForWriting(AvroSchemaRegistryClientMessageConverter.java:242)
      at org.springframework.cloud.stream.schema.avro.AbstractAvroMessageConverter.convertToInternal(AbstractAvroMessageConverter.java:174)
      at org.springframework.messaging.converter.AbstractMessageConverter.toMessage(AbstractMessageConverter.java:193)
      at org.springframework.cloud.stream.binding.MessageConverterConfigurer$ContentTypeConvertingInterceptor.preSend(MessageConverterConfigurer.java:253)
      at org.springframework.integration.channel.AbstractMessageChannel$ChannelInterceptorList.preSend(AbstractMessageChannel.java:538)
      at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:415)
      ... 31 more
  </code></pre>
  
  <p>Is the <code>register()</code> supposed to be creating a new schema if that one doesn't exist, or is the 404 appropriate behavior?  Quite confused.</p>