hi Jean-Baptiste, I'm sharing my pipeline with you: and I'm running it on Dataflow
public static PipelineResult run(Options options) { // Create the pipeline Pipeline pipeline = Pipeline.create(options); // Register the coder for pipeline FailsafeElementCoder<KV<String, String>, String> coder = FailsafeElementCoder.of( KvCoder.of( NullableCoder.of(StringUtf8Coder.of()), NullableCoder.of(StringUtf8Coder.of ())), NullableCoder.of(StringUtf8Coder.of())); CoderRegistry coderRegistry = pipeline.getCoderRegistry(); coderRegistry.registerCoderForType(coder.getEncodedTypeDescriptor(), coder); /* * Steps: * 1) Read messages in from Kafka * 2) Transform the messages into TableRows * - Transform message payload via UDF * - Convert UDF result to TableRow objects * 3) Write successful records out to BigQuery * 4) Write failed records out to BigQuery */ PCollection<RabbitMqMessage> messages = pipeline .apply("Read Rabbit queue", RabbitMqIO.read() .withUri("amqp://" + options.getRabbitUser() + ":" + options. getRabbitPassword() + "@" + options.getRabbitHost() + ":" + options. getRabbitPort()) .withQueue(options.getRabbitQueue()) ); final TupleTag<String> dummyTag = new TupleTag<String>(){}; PCollection<KV<String,String>> rowsProcessed = messages.apply("Transform RabbitMqMessage to Json", ParDo.of(new DoFn<RabbitMqMessage, KV<String,String>>() { @ProcessElement public void processElement(ProcessContext c) { String jsonInString = new String(c.element().getBody()); LOG.info(jsonInString); c.output(KV.of("test", jsonInString)); } })); PCollectionTuple convertedTableRows = rowsProcessed.apply("ConvertMessageToTableRow", new MessageToTableRow (options)); /* * Step #3: Write the successful records out to BigQuery */ WriteResult writeResult = convertedTableRows .get(TRANSFORM_OUT) .apply( "WriteSuccessfulRecords", BigQueryIO.writeTableRows() .withoutValidation() .withCreateDisposition(CreateDisposition.CREATE_NEVER) .withWriteDisposition(WriteDisposition.WRITE_APPEND) .withExtendedErrorInfo() .withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS) .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors()) .to(options.getOutputTableSpec())); /* * Step 3 Contd. * Elements that failed inserts into BigQuery are extracted and converted to FailsafeElement */ PCollection<FailsafeElement<String, String>> failedInserts = writeResult .getFailedInsertsWithErr() .apply( "WrapInsertionErrors", MapElements.into(FAILSAFE_ELEMENT_CODER.getEncodedTypeDescriptor()) .via(RabbitMqToBigQuery::wrapBigQueryInsertError)) .setCoder(FAILSAFE_ELEMENT_CODER); /* * Step #4: Write failed records out to BigQuery */ PCollectionList.of(convertedTableRows.get(UDF_DEADLETTER_OUT)) .and(convertedTableRows.get(TRANSFORM_DEADLETTER_OUT)) .apply("Flatten", Flatten.pCollections()) .apply( "WriteTransformationFailedRecords", WriteKafkaMessageErrors.newBuilder() .setErrorRecordsTable( ObjectUtils.firstNonNull( options.getOutputDeadletterTable(), options.getOutputTableSpec() + DEFAULT_DEADLETTER_TABLE_SUFFIX)) .setErrorRecordsTableSchema(SchemaUtils.DEADLETTER_SCHEMA) .build()); /* * Step #5: Insert records that failed BigQuery inserts into a deadletter table. */ failedInserts.apply( "WriteInsertionFailedRecords", ErrorConverters.WriteStringMessageErrors.newBuilder() .setErrorRecordsTable( ObjectUtils.firstNonNull( options.getOutputDeadletterTable(), options.getOutputTableSpec() + DEFAULT_DEADLETTER_TABLE_SUFFIX)) .setErrorRecordsTableSchema(SchemaUtils.DEADLETTER_SCHEMA) .build()); return pipeline.run(); } Em sáb., 16 de jan. de 2021 às 02:02, Jean-Baptiste Onofre <j...@nanthrax.net> escreveu: > Hi, > > Let me take a look. > > Regards > JB > > Le 15 janv. 2021 à 20:43, Rafael Ribeiro <rfribe...@gmail.com> a écrit : > > hi All, > > I'm facing the problem reported on > https://issues.apache.org/jira/browse/BEAM-6516 > > running a flex template based on kafka-to-bigquery and adapted to rabbitmq > > java.lang.RuntimeException: Exception while finalizing checkpoint > > 1. > 1. at org.apache.beam.runners.dataflow.worker. > StreamingModeExecutionContext.lambda$flushState$0 ( > StreamingModeExecutionContext.java:403 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=org%2Fapache.beam.runners.dataflow.worker%2FStreamingModeExecutionContext.java&line=403&project=rabbitmq-integration-301417> > ) > 2. at org.apache.beam.runners.dataflow.worker. > StreamingDataflowWorker.lambda$callFinalizeCallbacks$3 ( > StreamingDataflowWorker.java:1218 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=org%2Fapache.beam.runners.dataflow.worker%2FStreamingDataflowWorker.java&line=1218&project=rabbitmq-integration-301417> > ) > 3. at java.util.concurrent.ThreadPoolExecutor.runWorker ( > ThreadPoolExecutor.java:1149 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=java%2Futil.concurrent%2FThreadPoolExecutor.java&line=1149&project=rabbitmq-integration-301417> > ) > 4. at java.util.concurrent.ThreadPoolExecutor$Worker.run ( > ThreadPoolExecutor.java:624 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=java%2Futil.concurrent%2FThreadPoolExecutor.java&line=624&project=rabbitmq-integration-301417> > ) > 5. at java.lang.Thread.run (Thread.java:748 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=java%2Flang%2FThread.java&line=748&project=rabbitmq-integration-301417> > ) > > Caused by: java.io.IOException > > 1. > 1. at com.rabbitmq.client.impl.AMQChannel.wrap (AMQChannel.java:129 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FAMQChannel.java&line=129&project=rabbitmq-integration-301417> > ) > 2. at com.rabbitmq.client.impl.AMQChannel.wrap (AMQChannel.java:125 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FAMQChannel.java&line=125&project=rabbitmq-integration-301417> > ) > 3. at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc ( > AMQChannel.java:147 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FAMQChannel.java&line=147&project=rabbitmq-integration-301417> > ) > 4. at com.rabbitmq.client.impl.ChannelN.txCommit (ChannelN.java:1540 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FChannelN.java&line=1540&project=rabbitmq-integration-301417> > ) > 5. at com.rabbitmq.client.impl.recovery. > AutorecoveringChannel.txCommit (AutorecoveringChannel.java:663 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl.recovery%2FAutorecoveringChannel.java&line=663&project=rabbitmq-integration-301417> > ) > 2. > > Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; > protocol method: #method<channel.close>(reply-code=406, > reply-text=PRECONDITION_FAILED - unknown delivery tag 2, class-id=60, > method-id=80) > > 1. > 1. at com.rabbitmq.utility.ValueOrException.getValue ( > ValueOrException.java:66 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.utility%2FValueOrException.java&line=66&project=rabbitmq-integration-301417> > ) > 2. at com.rabbitmq.utility. > BlockingValueOrException.uninterruptibleGetValue ( > BlockingValueOrException.java:36 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.utility%2FBlockingValueOrException.java&line=36&project=rabbitmq-integration-301417> > ) > 3. at com.rabbitmq.client.impl. > AMQChannel$BlockingRpcContinuation.getReply (AMQChannel.java:502 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FAMQChannel.java&line=502&project=rabbitmq-integration-301417> > ) > 4. at com.rabbitmq.client.impl.AMQChannel.privateRpc ( > AMQChannel.java:293 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FAMQChannel.java&line=293&project=rabbitmq-integration-301417> > ) > 5. at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc ( > AMQChannel.java:141 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FAMQChannel.java&line=141&project=rabbitmq-integration-301417> > ) > > Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; > protocol method: #method<channel.close>(reply-code=406, > reply-text=PRECONDITION_FAILED - unknown delivery tag 2, class-id=60, > method-id=80) > > 1. > 1. at com.rabbitmq.client.impl.ChannelN.asyncShutdown ( > ChannelN.java:522 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FChannelN.java&line=522&project=rabbitmq-integration-301417> > ) > 2. at com.rabbitmq.client.impl.ChannelN.processAsync ( > ChannelN.java:346 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FChannelN.java&line=346&project=rabbitmq-integration-301417> > ) > 3. at com.rabbitmq.client.impl. > AMQChannel.handleCompleteInboundCommand (AMQChannel.java:182 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FAMQChannel.java&line=182&project=rabbitmq-integration-301417> > ) > 4. at com.rabbitmq.client.impl.AMQChannel.handleFrame ( > AMQChannel.java:114 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FAMQChannel.java&line=114&project=rabbitmq-integration-301417> > ) > 5. at com.rabbitmq.client.impl.AMQConnection.readFrame ( > AMQConnection.java:672 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FAMQConnection.java&line=672&project=rabbitmq-integration-301417> > ) > 6. at com.rabbitmq.client.impl.AMQConnection.access$300 ( > AMQConnection.java:48 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FAMQConnection.java&line=48&project=rabbitmq-integration-301417> > ) > 7. at com.rabbitmq.client.impl.AMQConnection$MainLoop.run ( > AMQConnection.java:599 > > <https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2021-01-14_12_12_16-8803828081371959712&file=com%2Frabbitmq.client.impl%2FAMQConnection.java&line=599&project=rabbitmq-integration-301417> > ) > > > When I'm receiving the messages from a Rabbitmq queue on Dataflow > > I already tried to add "use_deprecated_read" on the parameters list > > But I'm still seeing many error messages > > and sometimes the message is duplicated on BQ > > my rabbitmq queue configuration > Details > Type topic > Features > durable: true > Policy > > > > > > > > > Does anyone have any plans to fix this issue or any workaround to work? > > -- > Rafael Fernando Ribeiro > > > -- Rafael Fernando Ribeiro