https://flink.apache.org/ logo
Join Slack
Powered by
Hi all, I am trying to implement this concept "Con...
# troubleshooting
j
Hi all, I am trying to implement this concept "Confluent Avro Format" and I'm referring this page https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/table/formats/avro-confluent/. However, I'm facing few challenges, can someone help me here.
👀 1
my code to define source and sink table:
Copy code
def setup_user_created_source_table(t_env, cfg):
    te_user_created_source_ddl = """
                                create table %s(
                                    id string,
                                    name string,
                                    email string
                                ) WITH (
                                    'connector' = 'kafka',
                                    'property-version' = 'universal',
                                    'topic' = '%s',
                                    'properties.bootstrap.servers' = '%s',
                                    'properties.group.id' = '%s',
                                    'scan.startup.mode' = 'latest-offset',
                                    'format' = 'json'
                                )
                            """
    # register table and return it
    t_env.execute_sql(te_user_created_source_ddl %("user_created_source", "user_created_source_topic", "localhost:29092", "user_created_source_group_1"))
    user_created_source_table = t_env.from_path("user_created_source")
    return user_created_source_table

def setup_user_created_sink(st_env, cfg):    
    te_user_created_sink_ddl = """
                            CREATE TABLE %s(
                                the_kafka_key STRING,
                                id STRING,
                                name STRING, 
                                email STRING

                            ) WITH (

                            'connector' = 'kafka',
                            'topic' = '%s',
                            'properties.bootstrap.servers' = '%s',
                            'key.format' = 'raw',
                            'key.fields' = 'the_kafka_key',
                            'value.format' = 'avro-confluent',
                            'value.avro-confluent.url' = '<http://localhost:8082>',
                            'value.fields-include' = 'EXCEPT_KEY'
                            )
                                        """
    # register table and return it
    st_env.execute_sql(te_user_created_sink_ddl%("user_created_sink", "user_created_sink", "localhost:29092"))
    setup_user_created_sink_table = st_env.from_path("user_created_sink")
    return setup_user_created_sink_table
Flink Job code:
Copy code
import sys
from common import setup_user_created_source_table, setup_stream_table_environment ,setup_user_created_sink
 
PARALLELISM=1
 

def execute_join(env, st_env, cfg):
 
    user_created_source_table = setup_user_created_source_table(st_env, cfg) 
    user_created_sink_table = setup_user_created_sink(st_env, cfg)
      
    query = """
        SELECT id as the_kafka_key, id, name, email from user_created_source
    """  

    print(query)
    # sys.stdout.flush()

    user_created_source_table = st_env.sql_query(query)
    # user_created_source_table.execute().print()
    user_created_source_table.execute_insert("user_created_sink").wait()

def execute():
    print("EXECUTING USER CREATED")
    sys.stdout.flush()
    env, st_env, cfg = setup_stream_table_environment()
    st_env.get_config().set("pipeline.name", "user_created")
    env.set_parallelism(PARALLELISM)
    execute_join(env, st_env, cfg)

if __name__ == '__main__':
    execute()
Error:
Copy code
Caused by: 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(Thread.java:834)
Caused by: java.io.IOException: Failed to deserialize consumer record ConsumerRecord(topic = user_created_source_topic, partition = 0, leaderEpoch = 0, offset = 9, CreateTime = 1689822946536, serialized key size = -1, serialized value size = 59, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = [B@429303c6).
        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: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator
        at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:99)
        at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
        at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
        at org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask$AsyncDataOutputToOutput.emitRecord(SourceOperatorStreamTask.java:313)
        at org.apache.flink.streaming.api.operators.source.SourceOutputWithWatermarks.collect(SourceOutputWithWatermarks.java:110)
        at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter$SourceOutputWrapper.collect(KafkaRecordEmitter.java:67)
        at org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:84)
        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: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator
        at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:99)
        at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
        at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
        at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:56)
        at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:29)
        at StreamExecCalc$7.processElement(Unknown Source)
        at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
        ... 23 more
Caused by: java.lang.RuntimeException: Failed to serialize row.
        at org.apache.flink.formats.avro.AvroRowDataSerializationSchema.serialize(AvroRowDataSerializationSchema.java:90)
        at org.apache.flink.formats.avro.AvroRowDataSerializationSchema.serialize(AvroRowDataSerializationSchema.java:40)
        at org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaRecordSerializationSchema.serialize(DynamicKafkaRecordSerializationSchema.java:116)
        at org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaRecordSerializationSchema.serialize(DynamicKafkaRecordSerializationSchema.java:36)
        at org.apache.flink.connector.kafka.sink.KafkaWriter.write(KafkaWriter.java:196)
        at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.processElement(SinkWriterOperator.java:160)
        at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
        ... 29 more
Caused by: org.apache.flink.util.WrappingRuntimeException: Failed to serialize schema registry.
        at org.apache.flink.formats.avro.RegistryAvroSerializationSchema.serialize(RegistryAvroSerializationSchema.java:90)
        at org.apache.flink.formats.avro.AvroRowDataSerializationSchema.serialize(AvroRowDataSerializationSchema.java:88)
        ... 35 more
Caused by: java.net.ConnectException: Connection refused: connect
        at java.base/java.net.PlainSocketImpl.waitForConnect(Native Method)
        at java.base/java.net.PlainSocketImpl.socketConnect(PlainSocketImpl.java:107)
        at java.base/java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:412)
        at java.base/java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:255)
        at java.base/java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:237)
        at java.base/java.net.Socket.connect(Socket.java:608)
        at java.base/sun.net.NetworkClient.doConnect(NetworkClient.java:177)
        at java.base/sun.net.www.http.HttpClient.openServer(HttpClient.java:508)
        at java.base/sun.net.www.http.HttpClient.openServer(HttpClient.java:603)
        at java.base/sun.net.www.http.HttpClient.<init>(HttpClient.java:276)
        at java.base/sun.net.www.http.HttpClient.New(HttpClient.java:375)
        at java.base/sun.net.www.http.HttpClient.New(HttpClient.java:396)
        at java.base/sun.net.www.protocol.http.HttpURLConnection.getNewHttpClient(HttpURLConnection.java:1253)
        at java.base/sun.net.www.protocol.http.HttpURLConnection.plainConnect0(HttpURLConnection.java:1187)
        at java.base/sun.net.www.protocol.http.HttpURLConnection.plainConnect(HttpURLConnection.java:1081)
        at java.base/sun.net.www.protocol.http.HttpURLConnection.connect(HttpURLConnection.java:1015)
        at java.base/sun.net.www.protocol.http.HttpURLConnection.getOutputStream0(HttpURLConnection.java:1367)
        at java.base/sun.net.www.protocol.http.HttpURLConnection.getOutputStream(HttpURLConnection.java:1342)
        at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:265)
        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.registerSchema(RestService.java:507)
        at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:498)
        at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:471)
        at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.registerAndGetId(CachedSchemaRegistryClient.java:221)
        at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:283)
        at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:259)
        at org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.SchemaRegistryClient.register(SchemaRegistryClient.java:42)
        at org.apache.flink.formats.avro.registry.confluent.ConfluentSchemaRegistryCoder.writeSchema(ConfluentSchemaRegistryCoder.java:85)
        at org.apache.flink.formats.avro.RegistryAvroSerializationSchema.serialize(RegistryAvroSerializationSchema.java:85)
        ... 36 more
m
The stacktrace isn't about the serializing to avro-confluent, but it's about failing to deserialize the record from your input topic 
user_created_source_topic
j
Thank you @Martijn Visser for you reply. I don't think so I've a problem in deserialize the record from input topic 
user_created_source_topic
I produced this message 
b'{"id": "1", "name": "peter", "email": "<mailto:peter86@gmail.com|peter86@gmail.com>"}'
to the topic 
user_created_source_topic
and able to print in flink job. When I sink source topic table then I'm getting error.
m
Then it's a weird stack trace 😄
There's also 
Connection refused: connect
so can Flink actually connect to SR?
Especially if you run things in Docker, I've seen a lot of that stuff going wrong because of connectivity. Not sure if that's the case for you
j
what is SR
I'm using docker
m
Sorry, schema registry 🙂
If you use Docker, then you most likely can't point to localhost. You'll need to set the SCHEMA_REGISTRY_HOST_NAME and use that for connectivity. See https://github.com/MartijnVisser/flink-only-sql/blob/main/docker-compose.yml#L62-L65 on how I did that
j
Okay, I see what you are saying now.
so I need add this image to my docker-compose to make it work
Copy code
schema-registry:
    image: confluentinc/cp-schema-registry:7.2.1
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      - broker
    ports:
      - "8091:8091"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'
      SCHEMA_REGISTRY_LISTENERS: <http://0.0.0.0:8091>
Thank you @Martijn Visser it worked
🙌 1
Hey @Martijn Visser I got new problem after deploying in EKS cluster. Could you suggest me what might be the problem. py4j.protocol.Py4JJavaError: An error occurred while calling o171.executeInsert. : java.lang.NoSuchMethodError: 'org.apache.flink.avro.shaded.org.apache.avro.Schema org.apache.flink.formats.avro.typeutils.AvroSchemaConverter.convertToSchema(org.apache.flink.table.types.logical.LogicalType) ' at org.apache.flink.formats.avro.registry.confluent.RegistryAvroFormatFactory$2.createRuntimeEncoder(RegistryAvroFormatFactory.java:139) at org.apache.flink.formats.avro.registry.confluent.RegistryAvroFormatFactory$2.createRuntimeEncoder(RegistryAvroFormatFactory.java:130) at org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.createSerialization(KafkaDynamicSink.java:388) at org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.getSinkRuntimeProvider(KafkaDynamicSink.java:194) at org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecSink.createSinkTransformation(CommonExecSink.java:158) at org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink.translateToPlanInternal(StreamExecSink.java:176) at org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.translateToPlan(ExecNodeBase.java:161) at org.apache.flink.table.planner.delegation.StreamPlanner.$anonfun$translateToPlan$1(StreamPlanner.scala:85) at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233) at scala.collection.Iterator.foreach(Iterator.scala:937) at scala.collection.Iterator.foreach$(Iterator.scala:937) at scala.collection.AbstractIterator.foreach(Iterator.scala:1425) at scala.collection.IterableLike.foreach(IterableLike.scala:70) at scala.collection.IterableLike.foreach$(IterableLike.scala:69) at scala.collection.AbstractIterable.foreach(Iterable.scala:54) at scala.collection.TraversableLike.map(TraversableLike.scala:233) at scala.collection.TraversableLike.map$(TraversableLike.scala:226) at scala.collection.AbstractTraversable.map(Traversable.scala:104) at org.apache.flink.table.planner.delegation.StreamPlanner.translateToPlan(StreamPlanner.scala:84) at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:197) at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1803) at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:881) at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:989) at org.apache.flink.table.api.internal.TablePipelineImpl.execute(TablePipelineImpl.java:57) at org.apache.flink.table.api.Table.executeInsert(Table.java:1074) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) at java.base/java.lang.reflect.Method.invoke(Unknown Source) at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374) at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282) at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79) at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238) at java.base/java.lang.Thread.run(Unknown Source)
m
Looks like the cluster is missing the necessary artifacts
j
I added flink-sql-avro-confluent-registry-1.17.1.jar flink-avro-confluent-registry-1.17.1.jar flink-avro-1.17.1.jar
m
I think the avro-confluent one shouldn’t be there. And I’m not sure if there’s an equivalent say sql-avro. I’m not near a computer so can’t check it at the moment
🙌 1
j
You are right I took it out and it worked.
3 Views
Open in Slack
PreviousNext
Powered by