https://flink.apache.org/ logo
Join Slack
Powered by
Hi team, we encountered the schema forbidden issue...
# troubleshooting
l
Hi team, we encountered the schema forbidden issue during deployment with the changes of flink version upgrade (1.16.1 -> 1.17.1), hope to get some help here πŸ™. In the changes of upgrade flink version from 1.16.1 to 1.17.1, we also switched from our customized schema-registry dependency to 
org.apache.flink:flink-avro-confluent-registry:1.17.1
. Previously we never encountered this issue using our customized schema-registry, but after the changes, the error in the job was like this:
Copy code
97112e6_4e578e017158b6c826cf089a3d845171_10_5) switched from RUNNING to FAILED on sessionizer-job-taskmanager-1-1 @ 192.168.71.13 (dataPort=40879).
java.io.IOException: Could not register schema in registry
    at org.apache.flink.formats.avro.registry.confluent.ConfluentSchemaRegistryCoder.writeSchema(ConfluentSchemaRegistryCoder.java:90) ~[?:?]
    at org.apache.flink.formats.avro.RegistryAvroSerializationSchema.serialize(RegistryAvroSerializationSchema.java:85) ~[?:?]
    ...
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Forbidden; error code: 403
    at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:301) ~[?:?]
    at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:371) ~[?:?]
    ...
    at org.apache.flink.formats.avro.registry.confluent.ConfluentSchemaRegistryCoder.writeSchema(ConfluentSchemaRegistryCoder.java:85) ~[?:?]
And the job keeps switching from RUNNING to CANCELING to CANCELED. We checked the schema of the kafka topic we are producing to, it's owned by other teams, not us. And the job just kept auto registering this target schema, which cannot be disabled by setting 
auto.register.schemas
to 
false
, that's why it said Forbidden. We cannot change the owner or add another owner on that schema. With this limitation, is there any way to avoid this issue while switching to 1.17.1 with the 
org.apache.flink:flink-avro-confluent-registry:1.17.1
? Thanks! gratitude thank you
Below is the complete error from the job:
Copy code
97112e6_4e578e017158b6c826cf089a3d845171_10_5) switched from RUNNING to FAILED on sessionizer-job-taskmanager-1-1 @ 192.168.71.13 (dataPort=40879).
java.io.IOException: Could not register schema in registry
    at org.apache.flink.formats.avro.registry.confluent.ConfluentSchemaRegistryCoder.writeSchema(ConfluentSchemaRegistryCoder.java:90) ~[?:?]
    at org.apache.flink.formats.avro.RegistryAvroSerializationSchema.serialize(RegistryAvroSerializationSchema.java:85) ~[?:?]
    at app.serialization.CustomerSessionSerializationSchema.serialize(CustomerSessionSerializationSchema.java:63) ~[?:?]
    at app.serialization.CustomerSessionSerializationSchema.serialize(CustomerSessionSerializationSchema.java:18) ~[?:?]
    at org.apache.flink.connector.kafka.sink.KafkaWriter.write(KafkaWriter.java:197) ~[?:?]
    at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.processElement(SinkWriterOperator.java:158) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.api.operators.StreamFilter.processElement(StreamFilter.java:39) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.tasks.BroadcastingOutputCollector.collect(BroadcastingOutputCollector.java:77) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.tasks.BroadcastingOutputCollector.collect(BroadcastingOutputCollector.java:32) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:51) ~[flink-dist-1.17.1.jar:1.17.1]
    at app.types.CustomerSessionManager.closeSession(CustomerSessionManager.java:449) ~[?:?]
    at app.types.OpenSessionState.handleInactivityTimer(OpenSessionState.java:154) ~[?:?]
    at app.types.CustomerSessionManager.handleInactivityTimer(CustomerSessionManager.java:144) ~[?:?]
    at app.types.CustomerSessionManager.processTimer(CustomerSessionManager.java:119) ~[?:?]
    at app.operators.CustomerSessionProcessFunction.onTimer(CustomerSessionProcessFunction.java:147) ~[?:?]
    at org.apache.flink.streaming.api.operators.KeyedProcessOperator.invokeUserFunction(KeyedProcessOperator.java:91) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.api.operators.KeyedProcessOperator.onEventTime(KeyedProcessOperator.java:70) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.api.operators.InternalTimerServiceImpl.advanceWatermark(InternalTimerServiceImpl.java:302) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.api.operators.InternalTimeServiceManagerImpl.advanceWatermark(InternalTimeServiceManagerImpl.java:180) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.api.operators.AbstractStreamOperator.processWatermark(AbstractStreamOperator.java:602) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitWatermark(OneInputStreamTask.java:243) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.watermarkstatus.StatusWatermarkValve.findAndOutputNewMinWatermarkAcrossAlignedChannels(StatusWatermarkValve.java:200) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.watermarkstatus.StatusWatermarkValve.inputWatermark(StatusWatermarkValve.java:115) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:148) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:110) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:550) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:839) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:788) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:931) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745) ~[flink-dist-1.17.1.jar:1.17.1]
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) ~[flink-dist-1.17.1.jar:1.17.1]
    at java.lang.Thread.run(Unknown Source) ~[?:?]
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Forbidden; error code: 403
    at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:301) ~[?:?]
    at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:371) ~[?:?]
    at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:548) ~[?:?]
    at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:536) ~[?:?]
    at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:494) ~[?:?]
    at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.registerAndGetId(CachedSchemaRegistryClient.java:274) ~[?:?]
    at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:381) ~[?:?]
    at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:360) ~[?:?]
    at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:348) ~[?:?]
    at io.confluent.kafka.schemaregistry.client.SchemaRegistryClient.register(SchemaRegistryClient.java:47) ~[?:?]
    at org.apache.flink.formats.avro.registry.confluent.ConfluentSchemaRegistryCoder.writeSchema(ConfluentSchemaRegistryCoder.java:85) ~[?:?]
    ... 40 more
3 Views
Open in Slack
PreviousNext
Powered by