Gcsio migration (apache#29360)

* Cherry pick two previous commits on migrating gcs client The two commits are merged into one: * Reapply "Replace StorageV1 client with GCS client (apache#28079)" (apache#28721) * added project parameter to apiclient * Initialze storage client with project from pipeline option. --------- Co-authored-by: Bjorn Pedersen <[email protected]>
jto · Nov 17, 2023 · d0522c4 · d0522c4
1 parent 5e9ea74
commit d0522c4
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/.../flink/translation/wrappers/streaming/io/source/unbounded/FlinkUnboundedSourceReader.java b/.../flink/translation/wrappers/streaming/io/source/unbounded/FlinkUnboundedSourceReader.java
@@ -126,6 +126,9 @@ public InputStatus pollNext(ReaderOutput<WindowedValue<ValueWithRecordId<T>>> ou
     if (reader != null) {
       emitRecord(reader, output);
       return InputStatus.MORE_AVAILABLE;
+    } else if (noMoreSplits()) {
+      LOG.trace("No more splits.");
+      return InputStatus.END_OF_INPUT;
     } else {
       LOG.trace("No data available for now.");
       return InputStatus.NOTHING_AVAILABLE;