[Bug]: FirestoreIO writes fail: timer is set to max time (294247-01-09T04:00:54.775Z)

[rulle-io]

Apache Beam SDK for Java 2.46.0

What happened?

Caused by: java.lang.IllegalArgumentException: Cannot output with timestamp 2023-06-15T07:19:21.297Z. 
Output timestamps must be no earlier than the timestamp of the current input 
or timer (294247-01-09T04:00:54.775Z) minus the allowed skew (0 milliseconds) 
and no later than 294247-01-10T04:00:54.775Z. 
See the DoFn#getAllowedTimestampSkew() Javadoc for details on changing the allowed skew.
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.checkTimestamp(SimpleDoFnRunner.java:259)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.access$1300(SimpleDoFnRunner.java:85)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.outputWithTimestamp(SimpleDoFnRunner.java:416)
	org.apache.beam.sdk.io.gcp.firestore.FirestoreV1WriteFn$BaseBatchWriteFn$ProcessContextAdapter.output(FirestoreV1WriteFn.java:557)
	org.apache.beam.sdk.io.gcp.firestore.FirestoreV1WriteFn$BatchWriteFnWithSummary.handleWriteSummary(FirestoreV1WriteFn.java:96)
	org.apache.beam.sdk.io.gcp.firestore.FirestoreV1WriteFn$BaseBatchWriteFn.doFlush(FirestoreV1WriteFn.java:432)
	org.apache.beam.sdk.io.gcp.firestore.FirestoreV1WriteFn$BaseBatchWriteFn.flushBatch(FirestoreV1WriteFn.java:304)
	org.apache.beam.sdk.io.gcp.firestore.FirestoreV1WriteFn$BaseBatchWriteFn.processElement(FirestoreV1WriteFn.java:256)

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[AndreiArion]

Hello,
We are experience the same issue with the latest Apache Beam version (Java 2.48.0)
However at least some of the writes are getting to Firestore.

Maybe related to #22188 ?

Thanks,
Andrei

[AndreiArion]

Our pipeline reads data from a Firestore collection, do some transformations and and write it back to Firestore.
The erreur occurs during the write to firestore phase but the cause seems to be that FirestoreV1 source generates bad timestamps for some elements.

A workaround that worked in our context (batch processing) was to change the timestamp for all elements :
myCollection.apply("Pardo to map to Write objects , ParDo.of( new MyWriteConverterDoFn() )) // Firestore Sink provides a bad timestamp(MAX_LONG), so we need to fix it .apply("Change timestamp of all elements", WithTimestamps.of((SerializableFunction<Write, org.joda.time.Instant>) input -> { assert input != null; return org.joda.time.Instant.now(); }).withAllowedTimestampSkew(new org.joda.time.Duration(Long.MAX_VALUE))) .apply("Write PiiUsers to Firestore", FirestoreIO.v1().write().batchWrite().withRpcQosOptions(rpcQosOptions).build())
Please see all the implications in the documentation of withAllowedTimestampSkew method as it might not work in your setting !

[djaneluz]

I'm facing the same error, my batch pipeline does as follows.

pipeline
        .apply("GetBigQueryInfo", BigQueryIO.readTableRows()
            .fromQuery(getDeltaQuery(projectId, firestoreStateDataset))
            .usingStandardSql())
       .apply("TableRowToWrite", ParDo.of(new TableRowToWriteFn(projectId, databaseId)))
       .apply(GroupByKey.create())
       .apply(Values.create())
       .apply(Flatten.iterables())
       .apply("WriteToFirestore", FirestoreIO.v1().write()
            .batchWrite()
            .withRpcQosOptions(rpcQosOptions)
            .build());

and the WriteToFirestore step fails wit the exception:

Error message from worker: java.lang.RuntimeException: org.apache.beam.sdk.util.UserCodeException: java.lang.IllegalArgumentException: Cannot output with timestamp 2023-11-23T20:50:36.820Z. Output timestamps must be no earlier than the timestamp of the current input or timer (294247-01-09T04:00:54.775Z) minus the allowed skew (0 milliseconds) and no later than 294247-01-10T04:00:54.775Z. See the DoFn#getAllowedTimestampSkew() Javadoc for details on changing the allowed skew.
	org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn$1.output(GroupAlsoByWindowsParDoFn.java:187)
	org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner$1.outputWindowedValue(GroupAlsoByWindowFnRunner.java:108)
	org.apache.beam.runners.dataflow.worker.util.BatchGroupAlsoByWindowViaIteratorsFn.processElement(BatchGroupAlsoByWindowViaIteratorsFn.java:128)
	org.apache.beam.runners.dataflow.worker.util.BatchGroupAlsoByWindowViaIteratorsFn.processElement(BatchGroupAlsoByWindowViaIteratorsFn.java:56)
	org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.invokeProcessElement(GroupAlsoByWindowFnRunner.java:121)
	org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.processElement(GroupAlsoByWindowFnRunner.java:73)
	org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn.processElement(GroupAlsoByWindowsParDoFn.java:117)
	org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:44)
	org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:54)
	org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.runReadLoop(ReadOperation.java:218)
	org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.start(ReadOperation.java:169)
	org.apache.beam.runners.dataflow.worker.util.common.worker.MapTaskExecutor.execute(MapTaskExecutor.java:83)
	org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.executeWork(BatchDataflowWorker.java:319)
	org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.doWork(BatchDataflowWorker.java:291)
	org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.getAndPerformWork(BatchDataflowWorker.java:221)
	org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.doWork(DataflowBatchWorkerHarness.java:147)
	org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.call(DataflowBatchWorkerHarness.java:127)
	org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.call(DataflowBatchWorkerHarness.java:114)
	java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	org.apache.beam.sdk.util.UnboundedScheduledExecutorService$ScheduledFutureTask.run(UnboundedScheduledExecutorService.java:163)
	java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.apache.beam.sdk.util.UserCodeException: java.lang.IllegalArgumentException: Cannot output with timestamp 2023-11-23T20:50:36.820Z. Output timestamps must be no earlier than the timestamp of the current input or timer (294247-01-09T04:00:54.775Z) minus the allowed skew (0 milliseconds) and no later than 294247-01-10T04:00:54.775Z. See the DoFn#getAllowedTimestampSkew() Javadoc for details on changing the allowed skew.
	org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:39)
	org.apache.beam.sdk.io.gcp.firestore.FirestoreV1WriteFn$BatchWriteFnWithSummary$DoFnInvoker.invokeProcessElement(Unknown Source)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:211)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:185)
	org.apache.beam.runners.dataflow.worker.SimpleParDoFn.processElement(SimpleParDoFn.java:340)
	org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:44)
	org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:54)
	org.apache.beam.runners.dataflow.worker.SimpleParDoFn$1.output(SimpleParDoFn.java:285)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.outputWindowedValue(SimpleDoFnRunner.java:275)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.access$900(SimpleDoFnRunner.java:85)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:423)
	org.apache.beam.sdk.transforms.DoFnOutputReceivers$WindowedContextOutputReceiver.output(DoFnOutputReceivers.java:76)
	org.apache.beam.sdk.transforms.FlatMapElements$3.processElement(FlatMapElements.java:169)
	org.apache.beam.sdk.transforms.FlatMapElements$3$DoFnInvoker.invokeProcessElement(Unknown Source)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:211)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:188)
	org.apache.beam.runners.dataflow.worker.SimpleParDoFn.processElement(SimpleParDoFn.java:340)
	org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:44)
	org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:54)
	org.apache.beam.runners.dataflow.worker.SimpleParDoFn$1.output(SimpleParDoFn.java:285)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.outputWindowedValue(SimpleDoFnRunner.java:275)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.access$900(SimpleDoFnRunner.java:85)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:423)
	org.apache.beam.sdk.transforms.DoFnOutputReceivers$WindowedContextOutputReceiver.output(DoFnOutputReceivers.java:76)
	org.apache.beam.sdk.transforms.MapElements$2.processElement(MapElements.java:151)
	org.apache.beam.sdk.transforms.MapElements$2$DoFnInvoker.invokeProcessElement(Unknown Source)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:211)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:188)
	org.apache.beam.runners.dataflow.worker.SimpleParDoFn.processElement(SimpleParDoFn.java:340)
	org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:44)
	org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:54)
	org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn$1.output(GroupAlsoByWindowsParDoFn.java:185)
	... 22 more
Caused by: java.lang.IllegalArgumentException: Cannot output with timestamp 2023-11-23T20:50:36.820Z. Output timestamps must be no earlier than the timestamp of the current input or timer (294247-01-09T04:00:54.775Z) minus the allowed skew (0 milliseconds) and no later than 294247-01-10T04:00:54.775Z. See the DoFn#getAllowedTimestampSkew() Javadoc for details on changing the allowed skew.
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.checkTimestamp(SimpleDoFnRunner.java:259)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.access$1300(SimpleDoFnRunner.java:85)
	org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.outputWithTimestamp(SimpleDoFnRunner.java:416)
	org.apache.beam.sdk.io.gcp.firestore.FirestoreV1WriteFn$BaseBatchWriteFn$ProcessContextAdapter.output(FirestoreV1WriteFn.java:557)
	org.apache.beam.sdk.io.gcp.firestore.FirestoreV1WriteFn$BatchWriteFnWithSummary.handleWriteSummary(FirestoreV1WriteFn.java:96)
	org.apache.beam.sdk.io.gcp.firestore.FirestoreV1WriteFn$BaseBatchWriteFn.doFlush(FirestoreV1WriteFn.java:432)
	org.apache.beam.sdk.io.gcp.firestore.FirestoreV1WriteFn$BaseBatchWriteFn.flushBatch(FirestoreV1WriteFn.java:304)
	org.apache.beam.sdk.io.gcp.firestore.FirestoreV1WriteFn$BaseBatchWriteFn.processElement(

I'm using BEAM 2.52.0

[djaneluz]

UPDATE

I followed the recommendation from this article https://medium.com/@erik.b.ferm/combining-bounded-pcollections-with-custom-timestamps-in-apache-beam-b0fd911b2ac and it worked.

My pipeline now is like this:

pipeline
        .apply("GetBigQueryInfo", BigQueryIO.readTableRows()
            .fromQuery(getDeltaQuery(projectId, firestoreStateDataset))
            .usingStandardSql())
        .apply("TableRowToWrite", ParDo.of(new TableRowToWriteFn(projectId, databaseId)))
        .apply("SetTimeStampCombiner", Window.<KV<String, Write>>into(new GlobalWindows())
            .withTimestampCombiner(TimestampCombiner.EARLIEST))
        .apply(GroupByKey.create())
        .apply(Values.create())
        .apply(Flatten.iterables())
        .apply("WriteToFirestore", FirestoreIO.v1().write()
            .batchWrite()
            .withRpcQosOptions(rpcQosOptions)
            .build());

So the solution was to explicitly add a GlobalWindow using .withTimestampCombiner(TimestampCombiner.EARLIEST) before the GroupByKey.

[stankiewicz]

.take_issue