Open wtrzcinski opened 4 years ago
benhag
commented
4 years ago We have the same problem: key and value schema are registered twice when the RecordNameStrategy is used. Once using the default TopicNameStrategy (e.g. <topic-name>-key and <topic-name>-value) and once using the configured RecordNameStrategy.
The first message therefore defines the schema for the whole topic and it's not possible to sent messages using a different schema to the same topic.
Any news on this?
noraab
commented
4 years ago @wtrzcinski did you receive any response or found any solution?
wtrzcinski
commented
4 years ago Hi, sorry for late reply. No, I didn't get any response. It was long time ago, and I don't remember exactly, but as far as I remember the workaround was as follows:
At the very beginning we tried to send the message:
{
'value_schema': ...,
"records": [...]
}And we expected that the endpoint would return id of the schema which we wanted to use later. But we met the issue I described above - generally this scenario seems to be not working for us. So, as workaround, on the client side for every schema we send directly a request to schema registry in order to create the schema and that request returns "schema_id" which we can use later in a kafka event request. And then we send the records with already existing schema ids:
{
'key_schema_id': ...,
'value_schema_id': ...,
'records': [...]
}Best Regards Wojciech Trzciński
benhag
commented
4 years ago Requesting the schema ids from the schema-registry and sending them instead of the schema to the rest-proxy works. Thanks for the workaround @wtrzcinski
cobolbaby
commented
4 months ago Has the v3 interface already supported this capability? Does v2 only support io.confluent.kafka.serializers.subject.TopicNameStrategy?
curl --location 'http://10.190.81.165:8082/topics/lwtest3' \
--header 'Content-Type: application/vnd.kafka.jsonschema.v2+json' \
--data-raw '{
"records": [
{
"value": {
"email": "cobolbaby@qq.com",
"favorite_color": "green",
"favorite_number": 17,
"name": "itc180012"
}
}
],
"value_schema_id": 13
}'
{
"error_code": 42207,
"message": "Error serializing message. Error when fetching schema version. subject = lwtest3-value, schema = {\"$schema\":\"http://json-schema.org/draft-07/schema#\",\"description\":\"A Confluent Kafka Python User\",\"properties\":{\"email\":{\"description\":\"User's email\",\"format\":\"email\",\"type\":\"string\"},\"favorite_color\":{\"description\":\"User's favorite color\",\"enum\":[\"red\",\"green\"],\"type\":\"string\"},\"favorite_number\":{\"description\":\"User's favorite number\",\"exclusiveMinimum\":0,\"type\":\"number\"},\"name\":{\"description\":\"User's name\",\"minLength\":1,\"pattern\":\"^itc\\\\d{6}$\",\"type\":\"string\"}},\"required\":[\"name\",\"favorite_number\",\"favorite_color\",\"email\"],\"title\":\"User\",\"type\":\"object\"}\nSchema not found io.confluent.rest.exceptions.RestNotFoundException: Schema not found\nio.confluent.rest.exceptions.RestNotFoundException: Schema not found\n\tat io.confluent.kafka.schemaregistry.rest.exceptions.Errors.schemaNotFoundException(Errors.java:121)\n\tat io.confluent.kafka.schemaregistry.rest.resources.SubjectsResource.lookUpSchemaUnderSubject(SubjectsResource.java:132)\n\tat jdk.internal.reflect.GeneratedMethodAccessor79.invoke(Unknown Source)\n\tat java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.base/java.lang.reflect.Method.invoke(Method.java:566)\n\tat org.glassfish.jersey.server.model.internal.ResourceMethodInvocationHandlerFactory.lambda$static$0(ResourceMethodInvocationHandlerFactory.java:52)\n\tat org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher$1.run(AbstractJavaResourceMethodDispatcher.java:134)\n\tat org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.invoke(AbstractJavaResourceMethodDispatcher.java:177)\n\tat org.glassfish.jersey.server.model.internal.JavaResourceMethodDispatcherProvider$VoidOutInvoker.doDispatch(JavaResourceMethodDispatcherProvider.java:159)\n\tat org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.dispatch(AbstractJavaResourceMethodDispatcher.java:81)\n\tat org.glassfish.jersey.server.model.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:475)\n\tat org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:397)\n\tat org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:81)\n\tat org.glassfish.jersey.server.ServerRuntime$1.run(ServerRuntime.java:255)\n\tat org.glassfish.jersey.internal.Errors$1.call(Errors.java:248)\n\tat org.glassfish.jersey.internal.Errors$1.call(Errors.java:244)\n\tat org.glassfish.jersey.internal.Errors.process(Errors.java:292)\n\tat org.glassfish.jersey.internal.Errors.process(Errors.java:274)\n\tat org.glassfish.jersey.internal.Errors.process(Errors.java:244)\n\tat org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:265)\n\tat org.glassfish.jersey.server.ServerRuntime.process(ServerRuntime.java:234)\n\tat org.glassfish.jersey.server.ApplicationHandler.handle(ApplicationHandler.java:684)\n\tat org.glassfish.jersey.servlet.WebComponent.serviceImpl(WebComponent.java:394)\n\tat org.glassfish.jersey.servlet.ServletContainer.serviceImpl(ServletContainer.java:378)\n\tat org.glassfish.jersey.servlet.ServletContainer.doFilter(ServletContainer.java:553)\n\tat org.glassfish.jersey.servlet.ServletContainer.doFilter(ServletContainer.java:494)\n\tat org.glassfish.jersey.servlet.ServletContainer.doFilter(ServletContainer.java:431)\n\tat org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)\n\tat org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1626)\n\tat org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:552)\n\tat org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233)\n\tat org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:1624)\n\tat org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:143)\n\tat org.eclipse.jetty.server.handler.RequestLogHandler.handle(RequestLogHandler.java:54)\n\tat org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127)\n\tat io.confluent.kafka.schemaregistry.rest.RequestIdHandler.handle(RequestIdHandler.java:51)\n\tat org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127)\n\tat org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:235)\n\tat org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1440)\n\tat org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:188)\n\tat org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:505)\n\tat org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:1594)\n\tat org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:186)\n\tat org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1355)\n\tat org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141)\n\tat org.eclipse.jetty.server.handler.HandlerCollection.handle(HandlerCollection.java:146)\n\tat org.eclipse.jetty.server.handler.HandlerCollection.handle(HandlerCollection.java:146)\n\tat org.eclipse.jetty.server.handler.StatisticsHandler.handle(StatisticsHandler.java:181)\n\tat org.eclipse.jetty.server.handler.ContextHandlerCollection.handle(ContextHandlerCollection.java:234)\n\tat org.eclipse.jetty.server.handler.gzip.GzipHandler.handle(GzipHandler.java:772)\n\tat org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127)\n\tat org.eclipse.jetty.server.Server.handle(Server.java:516)\n\tat org.eclipse.jetty.server.HttpChannel.lambda$handle$1(HttpChannel.java:487)\n\tat org.eclipse.jetty.server.HttpChannel.dispatch(HttpChannel.java:732)\n\tat org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:479)\n\tat org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:277)\n\tat org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:311)\n\tat org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:105)\n\tat org.eclipse.jetty.io.ChannelEndPoint$1.run(ChannelEndPoint.java:104)\n\tat org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:883)\n\tat org.eclipse.jetty.util.thread.QueuedThreadPool$Runner.run(QueuedThreadPool.java:1034)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\n; error code: 40403"
}curl --location 'http://10.190.81.165:8082/v3/clusters/6nzE5kIdQY-IaxDsadO30g/topics/lwtest3/records' \
--header 'Content-Type: application/json' \
--data-raw '{
"value": {
"subject_name_strategy": "RECORD_NAME",
"schema_id": 13,
"data": {
"email": "cobolbaby@qq.com",
"favorite_color": "green",
"favorite_number": 17,
"name": "itc180012"
}
}
}'
{
"error_code": 200,
"cluster_id": "6nzE5kIdQY-IaxDsadO30g",
"topic_name": "lwtest3",
"partition_id": 0,
"offset": 6,
"timestamp": "2024-04-13T06:25:21.330Z",
"value": {
"type": "JSONSCHEMA",
"subject": "User",
"schema_id": 13,
"schema_version": 5,
"size": 98
}
}
Fryie
commented
3 months ago @cobolbaby To me, it looks like kafka-rest previously supported setting key.subject.name.strategy and value.subject.name.strategy at a global level which would appear to have worked in conjunction with setting the schema IDs explicitly. It looks like this configuration was removed in newer versions of kafka-rest (I don't know when exactly). I don't know if there is any other way of getting v2 to work. In v3, you can set the subject name strategy in the payload, which appears to work.
At least this is what I found out after attempting a pretty major update of kafka-rest.
Hi, I try to send messages to the kafka through the kafka-rest proxy with custom avro schemas per record. I tried to setup proper SubjectNameStrategy using properties in /etc/kafka-rest/kafka-rest.properties configuration file:
And I met very strange behavior as if the kafka-rest proxy tried to register my schemas twice. Once under "topic-name-key", "topic-name-value" subject names and once under actual record names. As visible in following logs from schema-registry container:
For the next records, which have different avro schema, the registry complains with following logs:
And the kafka rest proxy with the following exception:
At a first glance the issue seems to be that your SchemaRestProducer doesn't understand the configured subject strategy but your KafkaAvroSerializer does.
Are you supporting this schema per record scenario and it is just a bug? (or you don't support this scenario and the "subject.name.strategy" configuration shouldn't be accepted by the KafkaAvroSerializer or KafkaAvroSerializer shouldn't auto register schema.... Edit: I'm just guessing, I don't know the root cause, but it doesn't work properly)
Best Regards Wojtek