https://flink.apache.org/ logo
Join Slack
Powered by
Hi All, Can someone help out with reading avro dat...
# troubleshooting
r
Hi All, Can someone help out with reading avro data from Confluent Cloud Kafka topics in Flink? I have specified the Avro-confluent format using Flink's Table API in the src DDLs
Copy code
src_ddl = """
CREATE TABLE source_kafka (
    viewtime INT,
    registertime BIGINT,
    pageid STRING
) WITH (
    'connector' = 'kafka',
    'topic' = 'pageviews',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'avro-confluent',
    'avro-confluent.schema-registry.url' = '<http://localhost:8081/>',
    'properties.group.id' = 'test-001',
    'properties.bootstrap.servers' = 'localhost:9092'
);"""
But I keep running into this error.
Copy code
java.io.IOException: Failed to deserialize consumer record due to
    at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:56)
    at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:33)
    at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:143)
    at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:385)
    at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
    at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:542)
    at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:831)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:780)
    at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935)
    at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:914)
    at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550)
    at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.io.IOException: Failed to deserialize consumer record ConsumerRecord(topic = pageviews, partition = 0, leaderEpoch = 0, offset = 5, CreateTime = 1673563915562, serialized key size = 2, serialized value size = 21, headers = RecordHeaders(headers = [RecordHeader(key = task.generation, value = [48]), RecordHeader(key = task.id, value = [48]), RecordHeader(key = current.iteration, value = [53])], isReadOnly = false), key = [B@4ee8961e, value = [B@1a9f96a).
    at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:57)
    at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:53)
    ... 14 more
Caused by: java.io.IOException: Failed to deserialize Avro record.
    at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:106)
    at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:46)
    at org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:82)
    at org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaDeserializationSchema.deserialize(DynamicKafkaDeserializationSchema.java:113)
    at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:54)
    ... 15 more
Caused by: java.lang.ArrayIndexOutOfBoundsException
Any insight on how to fix would be really helpful?
s
Off top of my head: does your Avro schema also have 3 fields: viewtime, registertime and pageid? AFAIK columns in the table definition must match the fields in the schema, you can’t skip anything.
r
Hi, thank you so much for your response. My topic schema is
Copy code
{
  "connect.name": "ksql.pageviews",
  "fields": [
    {
      "name": "viewtime",
      "type": "long"
    },
    {
      "name": "userid",
      "type": "string"
    },
    {
      "name": "pageid",
      "type": "string"
    }
  ],
  "name": "pageviews",
  "namespace": "ksql",
  "type": "record"
}
Is this what you meant by "Avro schema"?
This is the stream I created to write data to the topic.
Copy code
CREATE STREAM PAGEVIEWS (viewtime BIGINT, userid VARCHAR, pageid VARCHAR)
  WITH (kafka_topic='pageviews', value_format='avro', partitions=1);
s
Yes, exactly. It doesn’t seem to match your table definition:
Copy code
viewtime INT,
registertime BIGINT,
pageid STRING
r
thank you so much, I have been looking at this for a while and I didn't notice the schema difference.
I've corrected my table definition now. But the error still persists. New table definition:
Copy code
from pyflink.common import Configuration
from pyflink.table import TableEnvironment, EnvironmentSettings
t_env = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())
t_env.get_config().get_configuration().set_string("execution.checkpointing.interval", "5s")

drop_src_ddl = "DROP TABLE IF EXISTS source_kafka;"
src_ddl = """
CREATE TABLE source_kafka (
    viewtime BIGINT,
    userid STRING,
    pageid STRING
) WITH (
    'connector' = 'kafka',
    'topic' = 'pageviews',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'avro-confluent',
    'avro-confluent.schema-registry.url' = '<http://localhost:8081/>',
    'properties.group.id' = 'test-001',
    'properties.bootstrap.servers' = 'broker:29092',
    'properties.zookeeper.connect' = 'zookeeper:2181'
);
"""
drop_sink_ddl = "DROP TABLE IF EXISTS sink_csv;"
sink_ddl = """
CREATE TABLE sink_csv (
    viewtime BIGINT,
    userid STRING,
    pageid STRING
) WITH (
    'connector' = 'filesystem',
        'format' = 'csv',
        'path' = './output',
        'sink.rolling-policy.rollover-interval' = '5s'
);
"""

# drop_sink_ddl = "DROP TABLE IF EXISTS sink_s3;"
# sink_ddl = """
# CREATE TABLE sink_s3 (
#     viewtime INT,
#     registertime BIGINT,
#     pageid STRING
# ) WITH (
#     'connector' = 'filesystem',
#         'format' = 'csv',
#         'path' = '<https://thescore-test.s3.amazonaws.com/test-flink/>',
#         'sink.rolling-policy.rollover-interval' = '5s'
# );
# """

t_env.execute_sql(drop_src_ddl)
t_env.execute_sql(src_ddl)
t_env.execute_sql(drop_sink_ddl)
t_env.execute_sql(sink_ddl)


# run a single INSERT query on the registered source table and emit the result to registered sink table
table_result1 = t_env.execute_sql("""
    INSERT INTO sink_csv
        SELECT viewtime,
    userid,
    pageid
    FROM source_kafka
""")
# get
s
Any chance you can post the full stacktrace? It’s hard to see where ArrayIndexOutOfBoundsException is coming from.
r
yes definitely
This is the job manager log
Copy code
java.io.IOException: Failed to deserialize consumer record due to
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:56) ~[flink-connector-kafka-1.16.0.jar:1.16.0]
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:33) ~[flink-connector-kafka-1.16.0.jar:1.16.0]
	at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:143) ~[flink-connector-files-1.16.0.jar:1.16.0]
	at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:385) ~[flink-streaming-java-1.16.0.jar:1.16.0]
	at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68) ~[flink-streaming-java-1.16.0.jar:1.16.0]
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) ~[flink-streaming-java-1.16.0.jar:1.16.0]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:542) ~[flink-streaming-java-1.16.0.jar:1.16.0]
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231) ~[flink-streaming-java-1.16.0.jar:1.16.0]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:831) ~[flink-streaming-java-1.16.0.jar:1.16.0]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:780) ~[flink-streaming-java-1.16.0.jar:1.16.0]
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935) ~[flink-dist-1.16.0.jar:1.16.0]
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:914) ~[flink-dist-1.16.0.jar:1.16.0]
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728) ~[flink-dist-1.16.0.jar:1.16.0]
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550) ~[flink-dist-1.16.0.jar:1.16.0]
	at java.lang.Thread.run(Unknown Source) ~[?:?]
Caused by: java.io.IOException: Failed to deserialize consumer record ConsumerRecord(topic = pageviews, partition = 0, leaderEpoch = 0, offset = 0, CreateTime = 1673976668209, serialized key size = 1, serialized value size = 21, headers = RecordHeaders(headers = [RecordHeader(key = task.generation, value = [48]), RecordHeader(key = task.id, value = [48]), RecordHeader(key = current.iteration, value = [48])], isReadOnly = false), key = [B@1befd7c0, value = [B@4ac1e760).
	at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:57) ~[flink-connector-kafka-1.16.0.jar:1.16.0]
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:53) ~[flink-connector-kafka-1.16.0.jar:1.16.0]
	... 14 more
Caused by: java.io.IOException: Failed to deserialize Avro record.
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:106) ~[flink-sql-avro-1.16.0.jar:1.16.0]
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:46) ~[flink-sql-avro-1.16.0.jar:1.16.0]
	at org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:82) ~[flink-dist-1.16.0.jar:1.16.0]
	at org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaDeserializationSchema.deserialize(DynamicKafkaDeserializationSchema.java:113) ~[flink-connector-kafka-1.16.0.jar:1.16.0]
	at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:54) ~[flink-connector-kafka-1.16.0.jar:1.16.0]
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:53) ~[flink-connector-kafka-1.16.0.jar:1.16.0]
	... 14 more
Caused by: java.net.ConnectException: Connection refused (Connection refused)
	at java.net.PlainSocketImpl.socketConnect(Native Method) ~[?:?]
	at java.net.AbstractPlainSocketImpl.doConnect(Unknown Source) ~[?:?]
	at java.net.AbstractPlainSocketImpl.connectToAddress(Unknown Source) ~[?:?]
	at java.net.AbstractPlainSocketImpl.connect(Unknown Source) ~[?:?]
	at java.net.Socket.connect(Unknown Source) ~[?:?]
	at sun.net.NetworkClient.doConnect(Unknown Source) ~[?:?]
	at sun.net.www.http.HttpClient.openServer(Unknown Source) ~[?:?]
	at sun.net.www.http.HttpClient.openServer(Unknown Source) ~[?:?]
	at sun.net.www.http.HttpClient.<init>(Unknown Source) ~[?:?]
	at sun.net.www.http.HttpClient.New(Unknown Source) ~[?:?]
	at sun.net.www.http.HttpClient.New(Unknown Source) ~[?:?]
	at sun.net.www.protocol.http.HttpURLConnection.getNewHttpClient(Unknown Source) ~[?:?]
	at sun.net.www.protocol.http.HttpURLConnection.plainConnect0(Unknown Source) ~[?:?]
	at sun.net.www.protocol.http.HttpURLConnection.plainConnect(Unknown Source) ~[?:?]
	at sun.net.www.protocol.http.HttpURLConnection.connect(Unknown Source) ~[?:?]
	at sun.net.www.protocol.http.HttpURLConnection.getInputStream0(Unknown Source) ~[?:?]
	at sun.net.www.protocol.http.HttpURLConnection.getInputStream(Unknown Source) ~[?:?]
	at java.net.HttpURLConnection.getResponseCode(Unknown Source) ~[?:?]
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:274) ~[flink-sql-avro-confluent-registry-1.16.0.jar:1.16.0]
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:364) ~[flink-sql-avro-confluent-registry-1.16.0.jar:1.16.0]
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:716) ~[flink-sql-avro-confluent-registry-1.16.0.jar:1.16.0]
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:698) ~[flink-sql-avro-confluent-registry-1.16.0.jar:1.16.0]
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaByIdFromRegistry(CachedSchemaRegistryClient.java:232) ~[flink-sql-avro-confluent-registry-1.16.0.jar:1.16.0]
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaBySubjectAndId(CachedSchemaRegistryClient.java:306) ~[flink-sql-avro-confluent-registry-1.16.0.jar:1.16.0]
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaById(CachedSchemaRegistryClient.java:291) ~[flink-sql-avro-confluent-registry-1.16.0.jar:1.16.0]
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.SchemaRegistryClient.getById(SchemaRegistryClient.java:64) ~[flink-sql-avro-confluent-registry-1.16.0.jar:1.16.0]
	at org.apache.flink.formats.avro.registry.confluent.ConfluentSchemaRegistryCoder.readSchema(ConfluentSchemaRegistryCoder.java:74) ~[flink-sql-avro-confluent-registry-1.16.0.jar:1.16.0]
	at org.apache.flink.formats.avro.RegistryAvroDeserializationSchema.deserialize(RegistryAvroDeserializationSchema.java:73) ~[flink-sql-avro-1.16.0.jar:1.16.0]
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:103) ~[flink-sql-avro-1.16.0.jar:1.16.0]
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:46) ~[flink-sql-avro-1.16.0.jar:1.16.0]
	at org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:82) ~[flink-dist-1.16.0.jar:1.16.0]
	at org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaDeserializationSchema.deserialize(DynamicKafkaDeserializationSchema.java:113) ~[flink-connector-kafka-1.16.0.jar:1.16.0]
	at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:54) ~[flink-connector-kafka-1.16.0.jar:1.16.0]
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:53) ~[flink-connector-kafka-1.16.0.jar:1.16.0]
	... 14 more
2023-01-17 17:47:54,760 INFO  org.apache.flink.runtime.resourcemanager.slotmanager.DeclarativeSlotManager [] - Clearing resource requirements of job 81e0b84f6012c773317f08145a8ea859
2023-01-17 17:47:54,765 INFO  org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Removing registered reader after failure for subtask 0 (#418) of source Source: source_kafka[1].
2023-01-17 17:47:54,767 INFO  org.apache.flink.runtime.jobmaster.JobMaster                 [] - 1 tasks will be restarted to recover the failed task b77b5c5e4a4a796e1518221175962735_cbc357ccb763df2852fee8c4fc7d55f2_0_418.
2023-01-17 17:47:54,767 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Job insert-into_default_catalog.default_database.sink_csv (81e0b84f6012c773317f08145a8ea859) switched from state RUNNING to RESTARTING.
Full task manager log after changes. I noticed there's no 
ArrayIndexOutOfBoundsException
like the first stack trace I sent.
Copy code
2023-01-17 12:50:44
java.io.IOException: Failed to deserialize consumer record due to
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:56)
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:33)
	at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:143)
	at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:385)
	at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:542)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:831)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:780)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:914)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550)
	at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.io.IOException: Failed to deserialize consumer record ConsumerRecord(topic = pageviews, partition = 0, leaderEpoch = 0, offset = 0, CreateTime = 1673976668209, serialized key size = 1, serialized value size = 21, headers = RecordHeaders(headers = [RecordHeader(key = task.generation, value = [48]), RecordHeader(key = task.id, value = [48]), RecordHeader(key = current.iteration, value = [48])], isReadOnly = false), key = [B@19abd1d5, value = [B@67f4d6b).
	at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:57)
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:53)
	... 14 more
Caused by: java.io.IOException: Failed to deserialize Avro record.
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:106)
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:46)
	at org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:82)
	at org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaDeserializationSchema.deserialize(DynamicKafkaDeserializationSchema.java:113)
	at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:54)
	... 15 more
Caused by: java.net.ConnectException: Connection refused (Connection refused)
	at java.base/java.net.PlainSocketImpl.socketConnect(Native Method)
	at java.base/java.net.AbstractPlainSocketImpl.doConnect(Unknown Source)
	at java.base/java.net.AbstractPlainSocketImpl.connectToAddress(Unknown Source)
	at java.base/java.net.AbstractPlainSocketImpl.connect(Unknown Source)
	at java.base/java.net.Socket.connect(Unknown Source)
	at java.base/sun.net.NetworkClient.doConnect(Unknown Source)
	at java.base/sun.net.www.http.HttpClient.openServer(Unknown Source)
	at java.base/sun.net.www.http.HttpClient.openServer(Unknown Source)
	at java.base/sun.net.www.http.HttpClient.<init>(Unknown Source)
	at java.base/sun.net.www.http.HttpClient.New(Unknown Source)
	at java.base/sun.net.www.http.HttpClient.New(Unknown Source)
	at java.base/sun.net.www.protocol.http.HttpURLConnection.getNewHttpClient(Unknown Source)
	at java.base/sun.net.www.protocol.http.HttpURLConnection.plainConnect0(Unknown Source)
	at java.base/sun.net.www.protocol.http.HttpURLConnection.plainConnect(Unknown Source)
	at java.base/sun.net.www.protocol.http.HttpURLConnection.connect(Unknown Source)
	at java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream0(Unknown Source)
	at java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream(Unknown Source)
	at java.base/java.net.HttpURLConnection.getResponseCode(Unknown Source)
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:274)
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:364)
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:716)
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:698)
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaByIdFromRegistry(CachedSchemaRegistryClient.java:232)
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaBySubjectAndId(CachedSchemaRegistryClient.java:306)
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaById(CachedSchemaRegistryClient.java:291)
	at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.SchemaRegistryClient.getById(SchemaRegistryClient.java:64)
	at org.apache.flink.formats.avro.registry.confluent.ConfluentSchemaRegistryCoder.readSchema(ConfluentSchemaRegistryCoder.java:74)
	at org.apache.flink.formats.avro.RegistryAvroDeserializationSchema.deserialize(RegistryAvroDeserializationSchema.java:73)
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:103)
	... 19 more
s
Interesting, it looks like it can’t connect to the Schema Registry to get the schema
🙏 1
Could be auth-related.
r
So not a deserialization issue?
I'm using the confluent cloud quickstart, so I don't have any authentication setup.
3 Views
Open in Slack
PreviousNext
Powered by