[Bug]: Beam: exception reading Kafka Avro with Schema registry
- Dominant language
- Java
- Stars
- 1.5k
- Forks
- 476
- Avg merge
- 18h 33m
- Merged PRs (30d)
- 204
Description
### Apache Hop version?
2.4
### Java version?
11
### Operating system
Windows
### What happened?
Migration: https://issues.apache.org/jira/browse/HOP-4111
https://issues.apache.org/jira/browse/HOP-4112
**Consumer**
I was testing HOP pipelines on a local environment with Kubernetes. I generated a HOP fat-jar and deployed it on our flink-kubernetes-operator. As a schema registry we are using apicurio-registry-sql (https://github.com/Apicurio/apicurio-registry) that is fully compatible with Confluent schema registry API.
There was an exception in the task manager while using Beam Kafka Consume for reading Avro messages from Kafka using a schema registry.
Please see the attached example-kafka-avro-consume.hpl. I was running it with the real ip of apicurio-registry and verified that the URL worked.
The exception that I got in the task manager:
```
Caused by: java.lang.ClassNotFoundException: io.confluent.kafka.serializers.AbstractKafkaSchemaSerDe
at java.base/java.net.URLClassLoader.findClass(Unknown Source)
at java.base/java.lang.ClassLoader.loadClass(Unknown Source)
at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:64)
at org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:74)
at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:48)
at java.base/java.lang.ClassLoader.loadClass(Unknown Source)
```
**Producer**
I was testing HOP pipelines on a local environment with Kubernetes. I generated a HOP fat-jar and deployed it on our flink-kubernetes-operator. As a schema registry we are using apicurio-registry-sql (https://github.com/Apicurio/apicurio-registry) that is fully compatible with Confluent schema registry API.
There was an exception in the job manager while using Beam Kafka Produce for writing Avro messages to Kafka using a schema registry
Please see the attached example-kafka-avro-produce.hpl.
The exception that I got in the job manager:
```
Caused by: java.lang.NullPointerException
at org.apache.beam.sdk.coders.AvroCoder$AvroDeterminismChecker.checkIndexedRecord(AvroCoder.java:629)
at org.apache.beam.sdk.coders.AvroCoder$AvroDeterminismChecker.recurse(AvroCoder.java:497)
at org.apache.beam.sdk.coders.AvroCoder$AvroDeterminismChecker.check(AvroCoder.java:476)
at org.apache.beam.sdk.coders.AvroCoder.(AvroCoder.java:316)
at org.apache.beam.sdk.coders.AvroCoder.(AvroCoder.java:308)
at org.apache.beam.sdk.coders.AvroGenericCoder.(AvroGenericCoder.java:26)
at org.apache.beam.sdk.coders.AvroGenericCoder.of(AvroGenericCoder.java:30)
at org.apache.beam.sdk.coders.AvroCoder.of(AvroCoder.java:151)
at org.apache.hop.beam.transforms.kafka.BeamProduceMeta.handleTransform(BeamProduceMeta.java:149)
at org.apache.hop.beam.pipeline.HopPipelineMetaToBeamPipelineConverter.handleBeamOutputTransforms(HopPipelineMetaToBeamPipelineConverter.java:344)
at org.apache.hop.beam.pipeline.HopPipelineMetaToBeamPipelineConverter.createPipeline(HopPipelineMetaToBeamPipelineConverter.java:219)
```
Pipelines to reproduce
[Archive.zip](https://github.com/apache/hop/files/11002555/Archive.zip)
### Issue Priority
Priority: 2
### Issue Component
Component: Beam, Component: Transforms
Contributor guide
Research direction
Start with the attached Archive.zip pipelines and the stack traces, then inspect org.apache.hop.beam.transforms.kafka.BeamProduceMeta at line 149 and the Beam pipeline converter. Reproduce the consumer and producer failures in the stated Flink deployment; done means Avro messages can use the schema registry without the missing serializer class or AvroCoder exception.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka
- Domain
- data-engineering, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100