apache / apache/hop

[Bug]: Beam: exception reading Kafka Avro with Schema registry

Open
#2,675 0 comments 0 reactions 0 assignees View on GitHub
awaiting triage Beam P2 Transforms
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.