У меня есть прототип конвейера Apache Beam, где я пытаюсь читать данные из RabbitMQ, используя следующую конфигурацию
p.apply("read_from_rabbit", RabbitMqIO.read()
.withUri(options.getRabbitMQUri())
.withQueue(options.getRabbitMQQueue())
)
.apply("extract_json_data", MapElements.via(new RabbitMessageToKafkaMessage()))
, когда я пытаюсь его запустить, я всегда получаю
Exception in thread "main" java.lang.NoClassDefFoundError: com/rabbitmq/client/QueueingConsumer$Delivery
at java.lang.Class.getDeclaredConstructors0(Native Method)
at java.lang.Class.privateGetDeclaredConstructors(Class.java:2671)
at java.lang.Class.getDeclaredConstructors(Class.java:2020)
at java.io.ObjectStreamClass.computeDefaultSUID(ObjectStreamClass.java:1793)
at java.io.ObjectStreamClass.access$100(ObjectStreamClass.java:72)
at java.io.ObjectStreamClass$1.run(ObjectStreamClass.java:253)
at java.io.ObjectStreamClass$1.run(ObjectStreamClass.java:251)
at java.security.AccessController.doPrivileged(Native Method)
at java.io.ObjectStreamClass.getSerialVersionUID(ObjectStreamClass.java:250)
at java.io.ObjectStreamClass.writeNonProxy(ObjectStreamClass.java:735)
at java.io.ObjectOutputStream.writeClassDescriptor(ObjectOutputStream.java:668)
at java.io.ObjectOutputStream.writeNonProxyDesc(ObjectOutputStream.java:1282)
at java.io.ObjectOutputStream.writeClassDesc(ObjectOutputStream.java:1231)
at java.io.ObjectOutputStream.writeClass(ObjectOutputStream.java:1213)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1120)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
at org.apache.beam.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils.java:51)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.CoderTranslation.toCustomCoder(CoderTranslation.java:119)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.CoderTranslation.toProto(CoderTranslation.java:83)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.SdkComponents.registerCoder(SdkComponents.java:250)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.PCollectionTranslation.toProto(PCollectionTranslation.java:35)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.SdkComponents.registerPCollection(SdkComponents.java:205)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.PTransformTranslation.translateAppliedPTransform(PTransformTranslation.java:369)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation$ParDoTranslator.translate(ParDoTranslation.java:120)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.PTransformTranslation.toProto(PTransformTranslation.java:149)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.getParDoPayload(ParDoTranslation.java:651)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.isSplittable(ParDoTranslation.java:666)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.PTransformMatchers$6.matches(PTransformMatchers.java:269)
at org.apache.beam.sdk.Pipeline$2.visitPrimitiveTransform(Pipeline.java:280)
at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:665)
at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
at org.apache.beam.sdk.runners.TransformHierarchy$Node.access$600(TransformHierarchy.java:317)
at org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:251)
at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:458)
at org.apache.beam.sdk.Pipeline.replace(Pipeline.java:258)
at org.apache.beam.sdk.Pipeline.replaceAll(Pipeline.java:208)
at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:154)
at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:64)
at org.apache.beam.sdk.Pipeline.run(Pipeline.java:313)
at org.apache.beam.sdk.Pipeline.run(Pipeline.java:299)
at myCompany.myProject.RabbitToKafka.runTransformer(RabbitToKafka.java:54)
at myCompany.myProject.RabbitToKafka.main(RabbitToKafka.java:61)
Caused by: java.lang.ClassNotFoundException: com.rabbitmq.client.QueueingConsumer$Delivery
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 48 more
Насколько я понимаю, поскольку мой класс RabbitMessageToKafkaMessage читает сообщения RabbitMQ, эти сообщения, в свою очередь, содержат данные RabbitMQ, в частности информацию о доставке, введенную в конструктор: public RabbitMqMessage(String routingKey, QueueingConsumer.Delivery delivery) { Это нормально?
Если это так, как я могу преобразоватьmy RabbitMQMessage в KV во время операции чтения?
EDIT 1 возникает ошибка при запуске моего конвейера из Eclipse.
EDIT 2 Этот проект - проект maven, запущенный с использованием Eclipse.Все мои зависимости Apache Bean имеют версию 2.12.0 (последняя).
Мое дерево зависимостей выглядит следующим образом (по крайней мере часть, касающаяся RabbitMQ)
myCompany:myProject:jar:0.1.5-SNAPSHOT
+- org.apache.beam:beam-sdks-java-io-rabbitmq:jar:2.12.0:compile
| \- com.rabbitmq:amqp-client:jar:5.4.3:compile