add autoclosable for asyncresultset

Author: sakthivelmanii

google-cloud-spanner/src/main/java/com/google/cloud/spanner/AsyncResultSetImpl.java

@@ -489,8 +489,8 @@ public ApiFuture<Void> setCallback(Executor exec, ReadyCallback cb) {
       // Start to fetch data and buffer these.
       this.result = SettableApiFuture.create();
       this.state = State.STREAMING_INITIALIZED;
-      initiateProduceRows();
-      //      this.service.execute(new InitiateStreamingRunnable());
+      this.service.execute(new InitiateStreamingRunnable());
+//      initiateProduceRows();
       this.executor = MoreExecutors.newSequentialExecutor(Preconditions.checkNotNull(exec));
       this.callback = Preconditions.checkNotNull(cb);
       pausedLatch.countDown();

google-cloud-spanner/src/test/java/com/google/cloud/spanner/RetryOnInvalidatedSessionTest.java

@@ -1593,25 +1593,26 @@ private void asyncTransactionManager_readAsync(
           AsyncTransactionStep<Void, Long> count =
               context.then(
                   (transaction, ignored) -> {
-                    AsyncResultSet rs = fn.apply(transaction);
-                    ApiFuture<Void> fut =
-                        rs.setCallback(
-                            queryExecutor,
-                            resultSet -> {
-                              while (true) {
-                                switch (resultSet.tryNext()) {
-                                  case OK:
-                                    counter.incrementAndGet();
-                                    break;
-                                  case DONE:
-                                    return CallbackResponse.DONE;
-                                  case NOT_READY:
-                                    return CallbackResponse.CONTINUE;
+                    try(AsyncResultSet rs = fn.apply(transaction)) {
+                      ApiFuture<Void> fut =
+                          rs.setCallback(
+                              queryExecutor,
+                              resultSet -> {
+                                while (true) {
+                                  switch (resultSet.tryNext()) {
+                                    case OK:
+                                      counter.incrementAndGet();
+                                      break;
+                                    case DONE:
+                                      return CallbackResponse.DONE;
+                                    case NOT_READY:
+                                      return CallbackResponse.CONTINUE;
+                                  }
                                 }
-                              }
-                            });
-                    return ApiFutures.transform(
-                        fut, input -> counter.get(), MoreExecutors.directExecutor());
+                              });
+                      return ApiFutures.transform(
+                          fut, input -> counter.get(), MoreExecutors.directExecutor());
+                    }
                   },
                   executor);
           CommitTimestampFuture ts = count.commitAsync();