[BEAM-8504] Fix a bug related to zero-row responses

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryStorageStreamSource.java

@@ -233,19 +233,19 @@ private synchronized boolean readNextRecord() throws IOException {
         fractionConsumedFromCurrentResponse = getFractionConsumed(currentResponse);
 
         Preconditions.checkArgument(
-            totalRowCountFromCurrentResponse > 0L,
-            "Row count from current response (%s) must be greater than one.",
+            totalRowCountFromCurrentResponse >= 0L,
+            "Row count from current response (%s) must be greater than or equal to zero.",
             totalRowCountFromCurrentResponse);
         Preconditions.checkArgument(
             0f <= fractionConsumedFromCurrentResponse && fractionConsumedFromCurrentResponse <= 1f,
             "Fraction consumed from current response (%s) is not in the range [0.0, 1.0].",
             fractionConsumedFromCurrentResponse);
         Preconditions.checkArgument(
-            fractionConsumedFromPreviousResponse < fractionConsumedFromCurrentResponse,
-            "Fraction consumed from previous response (%s) is not less than fraction consumed "
-                + "from current response (%s).",
-            fractionConsumedFromPreviousResponse,
-            fractionConsumedFromCurrentResponse);
+            fractionConsumedFromPreviousResponse <= fractionConsumedFromCurrentResponse,
+            "Fraction consumed from the current response (%s) has to be larger than or equal to "
+                + "the fraction consumed from the previous response (%s).",
+            fractionConsumedFromCurrentResponse,
+            fractionConsumedFromPreviousResponse);
       }
 
       record = datumReader.read(record, decoder);

sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOStorageReadTest.java

@@ -861,6 +861,9 @@ public void testFractionConsumed() throws Exception {
             // N.B.: All floating point numbers used in this test can be represented without
             // a loss of precision.
             createResponse(AVRO_SCHEMA, records.subList(0, 2), 0.250),
+            // Some responses may contain zero results, so we must ensure that we can are resilient
+            // to such responses.
+            createResponse(AVRO_SCHEMA, Lists.newArrayList(), 0.250),
             createResponse(AVRO_SCHEMA, records.subList(2, 4), 0.500),
             createResponse(AVRO_SCHEMA, records.subList(4, 7), 0.875));