Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #1700 (2018-02-28 22:47:48)

The read rows timeout is now set in teh right place
ResponseQueueReader needs to have the proper setting for partial read timeout.

spollapally: LGTM

Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataGrpcClient.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataGrpcClient.java
index 2b9b38f0e..414d4ae2f 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataGrpcClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataGrpcClient.java
@@ -453,7 +453,7 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
 
     // Delegate all resumable operations to the scanner. It will request a non-resumable scanner
     // during operation.
-    final ResponseQueueReader reader = new ResponseQueueReader();
+    final ResponseQueueReader reader = new ResponseQueueReader(retryOptions.getReadPartialRowTimeoutMillis());
     RetryingReadRowsOperation operation = createReadRowsRetryListener(request, reader);
     operation.setResultObserver(new StreamObserver<ReadRowsResponse>(){
       @Override
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ResponseQueueReader.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ResponseQueueReader.java
index 115e12034..177c6df0b 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ResponseQueueReader.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ResponseQueueReader.java
@@ -54,9 +54,9 @@ public class ResponseQueueReader
     return firstResponseTimer;
   }
 
-  private final BlockingQueue<ResultQueueEntry<FlatRow>> resultQueue =
-      new LinkedBlockingQueue<>();
-  private AtomicBoolean completionMarkerFound = new AtomicBoolean(false);
+  private final BlockingQueue<ResultQueueEntry<FlatRow>> resultQueue =  new LinkedBlockingQueue<>();
+  private final long waitTimeMs;
+  private final AtomicBoolean completionMarkerFound = new AtomicBoolean(false);
   private boolean lastResponseProcessed = false;
   private Long startTime;
   private ClientCallStreamObserver<ReadRowsRequest> requestStream;
@@ -65,10 +65,11 @@ public class ResponseQueueReader
   /**
    * <p>Constructor for ResponseQueueReader.</p>
    */
-  public ResponseQueueReader() {
+  public ResponseQueueReader(long waitTimeMs) {
     if (BigtableClientMetrics.isEnabled(MetricLevel.Info)) {
       startTime = System.nanoTime();
     }
+    this.waitTimeMs = waitTimeMs;
   }
 
   @Override
@@ -128,7 +129,7 @@ public class ResponseQueueReader
   protected ResultQueueEntry<FlatRow> getNext() throws IOException {
     ResultQueueEntry<FlatRow> queueEntry;
     try {
-      queueEntry = resultQueue.poll(250, TimeUnit.MILLISECONDS);
+      queueEntry = resultQueue.poll(waitTimeMs, TimeUnit.MILLISECONDS);
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
       throw new IOException("Interrupted while waiting for next result", e);


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/ResponseQueueReaderTest.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/ResponseQueueReaderTest.java
index f4fb3bb36..8371ab1a3 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/ResponseQueueReaderTest.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/ResponseQueueReaderTest.java
@@ -57,7 +57,7 @@ public class ResponseQueueReaderTest {
   @Before
   public void setup() {
     MockitoAnnotations.initMocks(this);
-    underTest = new ResponseQueueReader();
+    underTest = new ResponseQueueReader(10000);
     underTest.beforeStart(mockClientCallStreamObserver);
   }
 
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java
index 359b45274..95bb5a00a 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java
@@ -194,7 +194,7 @@ public class RetryingReadRowsOperationTest {
 
   @Test
   public void testSingleResponseWithQueue() throws UnsupportedEncodingException {
-    ResponseQueueReader reader = new ResponseQueueReader();
+    ResponseQueueReader reader = new ResponseQueueReader(10000);
     RetryingReadRowsOperation underTest = createOperation(reader);
     start(underTest);
     ByteString key = ByteString.copyFrom("SomeKey", "UTF-8");