Hello, I'm trying to setup an Apache Beam Java pipeline that:
- Reads messages from Kafka - Calls an external Python transform - Writes the output to Kafka Before this, I tried simple pipelines without Kafka: for example, using "Create" to generate some test values in Java and then pass those to a dummy Python transform. That worked so far. This is an extract of the pipeline's code: public static void main(String[] args) { // ... pipeline .apply(KafkaIO.<String, String>read() .withBootstrapServers(servers) .withTopic(readTopic) .withKeyDeserializer(StringDeserializer.class) .withValueDeserializer(StringDeserializer.class) .withoutMetadata() ) .apply(new CrossLanguageTransform(options.getExpansionServiceURL())) .apply(KafkaIO.<String, String>write() .withBootstrapServers(servers) .withTopic(writeTopic) .withKeySerializer(StringSerializer.class) .withValueSerializer(StringSerializer.class) ); pipeline.run().waitUntilFinish(); } And this is the external transform wrapper: public class CrossLanguageTransform extends PTransform<PCollection<KV<String, String>>, PCollection<KV<String, String>>> { private static final String URN = "beam:transforms:xlang:pythonwordcount"; private static String expansionAddress; public CrossLanguageTransform(String expansionAddress) { this.expansionAddress = expansionAddress; } @Override public PCollection<KV<String, String>> expand(PCollection<KV<String, String>> pcoll) { return pcoll.apply( "PythonWordCount", External.of(URN, new byte [] {}, expansionAddress) ); } } After running with: mvn exec:java -Dexec.mainClass=org.apache.beam.examples.CrossLanguagePipeline -Pportable-runner -Dexec.args="--runner=PortableRunner --jobEndpoint=localhost:8099 --useExternal=true --expansionServiceURL=localhost:9097 --experiments=beam_fn_api" I get: KeyError: 'beam:coders:javasdk:0.1' It's not even reaching my Python's transform code. I can reproduce this without registering the Python's transform, for example: python -m apache_beam.runners.portability.expansion_service_test -p 9097 What am I doing wrong here? -- This e-mail and any attachments may contain information that is privileged, confidential, and/or exempt from disclosure under applicable law. If you are not the intended recipient, you are hereby notified that any disclosure, copying, distribution or use of any information contained herein is strictly prohibited. If you have received this transmission in error, please immediately notify the sender and destroy the original transmission and any attachments, whether in electronic or hard copy format, without reading or saving.