Schema-registry: update from 4.0.0 to 4.1.0 sink connect get: SerializationException: Error retrieving Avro schema for id 1

[leechenqi]

[2018-05-04 10:52:42,218] ERROR WorkerSinkTask{id=connect-cloud-solr-sink-14} Task threw an uncaught and unrecoverable exception(org.apache.kafka.connect.runtime.WorkerTask:172)
org.apache.kafka.connect.errors.DataException: TWEETS
at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:95)
at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:468)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:301)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:205)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:173)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
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.common.errors.SerializationException: Error retrieving Avro schema for id 1
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Schema not found; error code: 40403
at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:202)
at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:229)
at io.confluent.kafka.schemaregistry.client.rest.RestService.lookUpSubjectVersion(RestService.java:296)
at io.confluent.kafka.schemaregistry.client.rest.RestService.lookUpSubjectVersion(RestService.java:284)

[leechenqi]

I think it‘s the same problem with #343 . It happens again. Connect (4.0.0) and registry (4.1.0) work well together , but connect(4.1.0) and registry (4.1.0) throw the exception.

[OneCricketeer]

Did the schema registry topic (or, more specifically, the brokers) change between upgrades as well?

What subjects are in your registry? What schema versions _are listed_ for that subject of your topic?

[eulerwang]

I have the same problem . all in confluent 4.1 ; when I want to sink one ksql created table , I have this problem .

[eulerwang]

curl -X POST -H "Content-Type: application/json" http://localhost:8083/connectors --data '{
"name": "XXXCCCVVV",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"connection.url": "jdbc:mysql://IP_OF_MYSQL:3306/XC",
"connection.user":"UUUUU",
"connection.password":"PPPPPPP",
"key.converter": "io.confluent.connect.avro.AvroConverter",
"key.converter.schema.registry.url": "http://ip_of_localhost:8081",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url": "http://ip_of_localhost:8081",
"topics":"XXXXXYYYY",
"pk.mode":"record_value",
"pf.fields":"ad_id",
"insert.mode":"upsert",
"auto.create":"true",
"auto.evolve":"true"
}
}' | jq

[anish-kunduru]

Ditto here. Using oss 4.1.0 for kafkarest, kafkaconnect, and schemaregistry causes this issue to be revealed in kafkaconnect when a confluent JDBC sink connector with multiple topics uses the same schema ID (caching enabled).

Rolling back the images to oss 4.0.0 immediately resolved the problem.

We are still on 0.11 brokers.

[leechenqi]

@cricket007 thanks for your reply.
The schema registry topic doesn't change at all. We just upgrade the confluent, schema registry and connect. No change with data, connectors and applications.
There are four subjects with the same schema id in our schema registry. When the source connector (version 4.1.0) writes to the four topics simultaneously (multiple topic with the same schema id), the early mentioned Exception occurs. But if just writes to a topic (just one topic with the schema id), it works well.
We think the bug discussed in #343 comes back.

[OneCricketeer]

I'm curious if anyone has tried a git diff between 4.0 and 4.1.0 to find out what may have changed? Also - anyone able to try the recent 4.1.1 release?

[OneCricketeer]

I know that one of schema registries I'm running has the schema "string" being used as the key for various topics - that is people are registering topics with Avro serializers but only using String keys, and it has generated ID 1 because it was the first schema ever registered... In other words, on a brand new registry and producer request, the Key would get ID 1, the value would get ID 2.

My point is that when connect deserializes the messages with that given configuration using AvroConverter on the key, it's going to use the Schema Registry to lookup both the key schema ID as well as the value. I've seen such errors as yours during a connect misconfiguration - the topic key is not actually Avro, but the key in bytes starts with what the AvroConverter thinks is (magic byte + schema ID)

And the error you're getting is that none of the versions listed under /subjects/topicName-key/versions/ contain schema ID 1

If i understand correctly, KSQL keys can only be strings and Kafka Connect only respects the message values to generate columns (for JDBC Connect).

Therefore, can you please try to use

key.converter=org.apache.kafka.connect.storage.StringConverter

I have been considering adding a PR to know if the schema for the error you see is being looked up as the key or the value... Just haven't found the time :/

---

As an aside: A thought of mine is that an Avro Schema of only a string takes more space in bytes than only using the String SerDe (i.e. skipping the registry) because you need to keep the magic byte + schema ID in the byte array along with the UTF8 string content - and if I understand correctly, there is no possible schema evolution you can apply to the avro schema of "string" (can't add, remove, or change a field).

[leechenqi]

@cricket007 We always use key.converter=org.apache.kafka.connect.storage.StringConverter in our connect.properties. The problem has nothing with the converter.

[OneCricketeer]

Okay, @leechenqi - mostly replying to @eulerwang 's configuration. Plus, I was saying I've seen a similar error (on Connect 3.3.0) when we've configured AvroConverter for keys that are actually strings.

Since your converter is only on values, that does isolate the issue.

The problem has nothing with the converter

It does, though. At least, the SchemaRegistryClient class used by the Avro Converter & Deserializer classes.

My suggestion here would be to setup remote debugging for your connect worker, then update your findings here.

This section of the connect-distributed startup script can be consulted for the important debug properties

# Set Debug options if enabled
if [ "x$KAFKA_DEBUG" != "x" ]; then

    # Use default ports
    DEFAULT_JAVA_DEBUG_PORT="5005"

In other words, you set export KAFKA_DEBUG=y, then run the connect task, then use your favorite IDE to setup remote debugging on port 5005

[jtuberville]

We are rolling back to 4.0.1 as we experienced this issue with the hdfs, s3, and jdbc sink connectors against 4.0.1 version of the schema registry.

[mageshn]

@leechenqi can you provide details about the Topic Name, the subjects that you are using? How many SR nodes do you have? Does the subject have the schema id registered? From what I can see its failing for lookUpSubjectVersion. This could happen if the schema was not registered against the subject.

[sgerrand]

We've also encountered this issue with the Confluent platform products (Kafka, Kafka Connect and Schema Registry) at version 4.1.0 and 4.1.1.

org.apache.kafka.connect.errors.DataException: **********
        at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:95)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:468)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:301)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:205)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:173)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
        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.common.errors.SerializationException: Error retrieving Avro schema for id 4511
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Schema not found io.confluent.rest.exceptions.RestNotFoundException: Schema not found
io.confluent.rest.exceptions.RestNotFoundException: Schema not found
        at io.confluent.kafka.schemaregistry.rest.exceptions.Errors.schemaNotFoundException(Errors.java:58)
        at io.confluent.kafka.schemaregistry.rest.resources.SubjectsResource.lookUpSchemaUnderSubject(SubjectsResource.java:89)
        at sun.reflect.GeneratedMethodAccessor25.invoke(Unknown Source)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.glassfish.jersey.server.model.internal.ResourceMethodInvocationHandlerFactory$1.invoke(ResourceMethodInvocationHandlerFactory.java:81)
        at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher$1.run(AbstractJavaResourceMethodDispatcher.java:144)
        at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.invoke(AbstractJavaResourceMethodDispatcher.java:161)
        at org.glassfish.jersey.server.model.internal.JavaResourceMethodDispatcherProvider$VoidOutInvoker.doDispatch(JavaResourceMethodDispatcherProvider.java:143)
        at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.dispatch(AbstractJavaResourceMethodDispatcher.java:99)
        at org.glassfish.jersey.server.model.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:389)
        at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:347)
        at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:102)
        at org.glassfish.jersey.server.ServerRuntime$2.run(ServerRuntime.java:326)
        at org.glassfish.jersey.internal.Errors$1.call(Errors.java:271)
        at org.glassfish.jersey.internal.Errors$1.call(Errors.java:267)
        at org.glassfish.jersey.internal.Errors.process(Errors.java:315)
        at org.glassfish.jersey.internal.Errors.process(Errors.java:297)
        at org.glassfish.jersey.internal.Errors.process(Errors.java:267)
        at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:317)
        at org.glassfish.jersey.server.ServerRuntime.process(ServerRuntime.java:305)
        at org.glassfish.jersey.server.ApplicationHandler.handle(ApplicationHandler.java:1154)
        at org.glassfish.jersey.servlet.WebComponent.serviceImpl(WebComponent.java:473)
        at org.glassfish.jersey.servlet.ServletContainer.serviceImpl(ServletContainer.java:408)
        at org.glassfish.jersey.servlet.ServletContainer.doFilter(ServletContainer.java:583)
        at org.glassfish.jersey.servlet.ServletContainer.doFilter(ServletContainer.java:524)
        at org.glassfish.jersey.servlet.ServletContainer.doFilter(ServletContainer.java:461)
        at org.eclipse.jetty.servlet.ServletHandler$CachedChain.doFilter(ServletHandler.java:1652)
        at org.eclipse.jetty.servlets.CrossOriginFilter.handle(CrossOriginFilter.java:257)
        at org.eclipse.jetty.servlets.CrossOriginFilter.doFilter(CrossOriginFilter.java:220)
        at org.eclipse.jetty.servlet.ServletHandler$CachedChain.doFilter(ServletHandler.java:1652)
        at org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:585)
        at org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:221)
        at org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1127)
        at org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:515)
        at org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:185)
        at org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1061)
        at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141)
        at org.eclipse.jetty.server.handler.HandlerCollection.handle(HandlerCollection.java:110)
        at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:97)
        at org.eclipse.jetty.server.handler.StatisticsHandler.handle(StatisticsHandler.java:159)
        at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:97)
        at org.eclipse.jetty.server.Server.handle(Server.java:499)
        at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:311)
        at org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:258)
        at org.eclipse.jetty.io.AbstractConnection$2.run(AbstractConnection.java:544)
        at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:635)
        at org.eclipse.jetty.util.thread.QueuedThreadPool$3.run(QueuedThreadPool.java:555)
        at java.lang.Thread.run(Thread.java:748)\n; error code: 40403
        at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:202)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:229)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.lookUpSubjectVersion(RestService.java:296)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.lookUpSubjectVersion(RestService.java:284)
        at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getVersionFromRegistry(CachedSchemaRegistryClient.java:125)
        at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getVersion(CachedSchemaRegistryClient.java:236)
        at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:152)
        at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserializeWithSchemaAndVersion(AbstractKafkaAvroDeserializer.java:194)
        at io.confluent.connect.avro.AvroConverter$Deserializer.deserialize(AvroConverter.java:120)
        at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:83)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:468)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:301)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:205)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:173)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
        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)\n

We're running 2 instances of Schema Registry.

[sgerrand]

📝 Downgrading Kafka Connect to version 4.0.1 but keeping the version of everything else at either 4.1.0 or 4.1.1 resolves this issue.

[jcasstevens]

@sgerrand Do you mind sharing how you were able to downgrade just Kafka Connect, and not the entire platform? I am fairly new to Kafka/Confluent..

[OneCricketeer]

@jcasstevens If you are not using Docker, then you might want to try downloading the Tarball of the older version of Confluent Platform, then only configuring the schema registry property file and running it via the provided script for it in the bin folder.

[sgerrand]

We run all the constituent parts of the Confluent platform as Docker
containers, with the notable exception of Kafka and ZooKeeper. That gives
us more fine grained control to mitigate errors like this one if and when
they surface.

Do you mind sharing how you were able to downgrade just Kafka Connect, and
not the entire platform? I am fairly new to Kafka/Confluent..

[thetoothpick]

We also ran into this bug when upgrading our cluster from 4.0.1 to 4.1.0. I dug into the code and I think I've found the cause.

We have two topics which use the same schema (let's call them topic-1 and topic-2). These topics are generated by the same code, but by two different producer processes. Somehow, in the schema registry, topic-1 is on version 16 and topic-2 is on version 17. The schemas are identical, and share a global schema ID (719).

I downloaded the source code and Kafka Connect in a debugger. I used the HDFS connector and subscribed to both topic-1 and topic-2. topic-1 was read correctly, but topic-2 generated an exception similar to the one in the original issue. The failed HTTP request was a POST to /subjects/<subject>?deleted=true, which according to the API documentation checks if a schema has already been registered.

The debugger showed that the first request for topic-1 did not have a schema.registry.schema.version property set in the schema, but that the request for topic-2 did, and it was set to 16 (the version of that schema in topic-1). Since the schema in question is registered at version 17 in topic-2, the schema registry returns 40403 and the connector crashes with the stack trace in the original issue.

Looking at the changes made between 4.0.1 and 4.1.0, I found fac6fc787f821a7cd5a2b3dc39be558ee28a1f86 (for PR #741), which changes how schemas are looked up. This change makes the schema lookup always use the global ID and ignore the subject name. I reverted the change locally and reran, and the connector successfully writes to HDFS.

So, it seems like that change caused a regression of #343.

[OneCricketeer]

Looking at that commit Sha, I think #838 rolled those changes partially back.

[thetoothpick]

839 (clone of #838) did change the behavior, and was released in 4.1.2. However, a new error is thrown in this version: Can't overwrite property: schema.registry.schema.version.

It looks like the new code looks up the schema twice, once by ID alone, then again by subject/ID. The second lookup gets the version number, and then the code tries to set that version number on the schema from the first lookup (which already has a version set).

Full stack trace:

[2018-07-26 15:14:00,904] ERROR WorkerSinkTask{id=qa12-book-db-external-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:172)
org.apache.kafka.connect.errors.DataException: topic-2
    at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:96)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:481)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:314)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:218)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:186)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
...
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 719
Caused by: org.apache.avro.AvroRuntimeException: Can't overwrite property: schema.registry.schema.version
    at org.apache.avro.JsonProperties.addProp(JsonProperties.java:187)
    at org.apache.avro.Schema.addProp(Schema.java:134)
    at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:165)
    at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserializeWithSchemaAndVersion(AbstractKafkaAvroDeserializer.java:195)
    at io.confluent.connect.avro.AvroConverter$Deserializer.deserialize(AvroConverter.java:132)
    at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:84)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:481)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:314)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:218)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:186)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
...

[mageshn]

@thetoothpick you are right that was missed in the original commit.

[mageshn]

https://github.com/confluentinc/schema-registry/pull/856 should fix it

[thetoothpick]

Thanks @mageshn. Is this going to be included in an upcoming release?

[sgerrand]

i.e. is it in the recently released 5.0.0 version? We're had to manage
patched instances for nearly two months to resolve this issue. I'd really
like it to be finally fixed with no future regressions.

[piraxis]

@sgerrand This is NOT fixed in 5.0.0

Also there is no known workaround for this other than patching it.

[thetoothpick]

It looks like #856 is now scheduled for 4.1.x.

[mageshn]

https://github.com/confluentinc/schema-registry/pull/856 fixes this issue. Its merged to 4.1.x, 5.0.x and also master.

[Evelyn-Bayes]

Leaving this comment hear for future reference as fixing this can be a touch confusing:

• The fix is in the jars produced by the Schema-Registry code.
• The fix is used by Schema-Registry clients, so they are actually downloaded onto the clients.
• For users running on docker the client images should contain the jars with the fix, so for example to fix the bug on connect you would want to download the 5.0.1 Connect docker image.
• For other users, they need to upgrade Schema-Registry and then force all clients to re-download the jars.

[OneCricketeer]

need to upgrade Schema-Registry and then force all clients to re-download the jars.

I've been able to use newer clients with an older server version. As long as the fix isn't a server-side change, then the REST API should all work the same, but having newer serializers will provide those fixes