title: Kafka connector and Schema Registry - Error Retrieving Avro Schema - Subject not found
body: <p>I have a topic that will eventually have lots of different schemas on it. For now it just has the one.
  I've created a connect job via REST like this:</p>
  
  <pre><code>{
   "name":"com.mycompany.sinks.GcsSinkConnector-auth2",
   "config": {
      "connector.class": "com.mycompany.sinks.GcsSinkConnector",
      "topics": "auth.events",
      "flush.size": 3,
      "my.setting":"bar",
      "key.converter":"org.apache.kafka.connect.storage.StringConverter",
      "key.deserializer":"org.apache.kafka.common.serialization.StringDerserializer",
      "value.converter":"io.confluent.connect.avro.AvroConverter",
      "value.converter.schema.registry.url":"http://schema-registry-service:8081",
      "value.subject.name.strategy":"io.confluent.kafka.serializers.subject.RecordNameStrategy",
      "group.id":"account-archiver"
  
   }
  }
  </code></pre>
  
  <p>I then push a message to that topic with a string key and an avro serialized payload. If I inspect the topic in the control center I see the correctly deserialized data coming through.
  Looking at the ouput from the connect instance though I see this in the logs</p>
  
  <pre><code>RROR WorkerSinkTask{id=com.mycompany.sinks.GcsSinkConnector-auth2-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask)
  org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
      at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178)
      at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)
      at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:487)
      at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:464)
      at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:320)
      at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:224)
      at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:192)
      at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
      at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
      at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at java.lang.Thread.run(Thread.java:748)
  Caused by: org.apache.kafka.connect.errors.DataException: Failed to deserialize data for topic auth.events to Avro:
      at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:107)
      at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:487)
      at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
      at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)
      ... 13 more
  Caused by: org.apache.kafka.common.errors.SerializationException: Error retrieving Avro schema for id 7
  Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Subject not found.; error code: 40401
      at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:226)
      at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:252)
      at io.confluent.kafka.schemaregistry.client.rest.RestService.lookUpSubjectVersion(RestService.java:319)
      at io.confluent.kafka.schemaregistry.client.rest.RestService.lookUpSubjectVersion(RestService.java:307)
      at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getVersionFromRegistry(CachedSchemaRegistryClient.java:158)
      at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getVersion(CachedSchemaRegistryClient.java:271)
      at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.schemaVersion(AbstractKafkaAvroDeserializer.java:184)
      at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:153)
      at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserializeWithSchemaAndVersion(AbstractKafkaAvroDeserializer.java:215)
      at io.confluent.connect.avro.AvroConverter$Deserializer.deserialize(AvroConverter.java:145)
      at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:90)
      at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:487)
      at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
      at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)
      at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)
      at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:487)
      at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:464)
      at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:320)
      at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:224)
      at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:192)
      at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
      at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
      at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at java.lang.Thread.run(Thread.java:748)
  </code></pre>
  
  <p>You can see from here that there are two related issues:</p>
  
  <ul>
  <li><code>Error retrieving Avro schema for id 7</code></li>
  <li><code>Subject not found.; error code: 40401</code></li>
  </ul>
  
  <p>What bugs me is that I've specified the strategy to be RecordNameStrategy which I think should use the magic byte to go and get the schema as opposed to the topic name, but it errors on Subject not found. I'm not sure if it's actually looking for a subject name or getting a schema by the ID.
  Either way by ssh-ing to the connect instance and doing a curl to 
  <code>http://schema-registry-service:8081/schemas/ids/7</code> I do get the schema returned.
  There is some additional logging above this stack trace which disappointingly looks like it's still using the wrong name strategy:</p>
  
  <pre><code>INFO AvroConverterConfig values:
      schema.registry.url = [http://schema-registry-service:8081]
      basic.auth.user.info = [hidden]
      auto.register.schemas = false
      max.schemas.per.subject = 1000
      basic.auth.credentials.source = URL
      schema.registry.basic.auth.user.info = [hidden]
      value.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
      key.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
  </code></pre>
  
  <p>Does anyone have any clues about how to resolve this? I'm using the following images:</p>
  
  <ul>
  <li>confluentinc/cp-kafka-connect:5.2.0</li>
  <li>confluentinc/cp-kafka:5.1.0</li>
  </ul>
  
  <p>Thanks</p>