Merge pull request #17 from topcoder-platform/PLAT-3536

Plat 3536

Author: eisbilir

src/main/java/com/topcoder/dal/DBAccessor.java

@@ -7,6 +7,7 @@
 import com.topcoder.dal.util.StreamJdbcTemplate;
 import com.topcoder.dal.util.ParameterizedExpression;
 
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import jdk.jshell.spi.ExecutionControl;
 import net.devh.boot.grpc.server.service.GrpcService;
@@ -317,6 +318,7 @@ public void query(QueryRequest request, StreamObserver<QueryResponse> responseOb
 
     @Override
     public StreamObserver<QueryRequest> streamQuery(StreamObserver<QueryResponse> responseObserver) {
+        logger.info("Stream started");
         return new StreamObserver<>() {
             Connection con = jdbcTemplate.getConnection();
             private final Duration streamTimeout = Duration.ofSeconds(20);
@@ -327,10 +329,11 @@ public StreamObserver<QueryRequest> streamQuery(StreamObserver<QueryResponse> re
 
             @Override
             public void onNext(QueryRequest request) {
-                resetStreamTimeout();
+                cancelStreamTimeout();
                 try {
                     QueryResponse response = executeQuery(request.getQuery(), con);
                     responseObserver.onNext(response);
+                    resetStreamTimeout();
                 } catch (Exception e) {
                     rollback();
                     cancelStreamTimeout();
@@ -377,7 +380,7 @@ private boolean debounce() {
 
             private boolean cancelStreamTimeout() {
                 ScheduledFuture<?> currentFuture = streamTimeoutFuture.get();
-                return currentFuture == null || currentFuture.cancel(false);
+                return currentFuture == null || currentFuture.isCancelled() || currentFuture.cancel(false);
             }
 
             private ScheduledFuture<?> scheduleStreamTimeout() {
@@ -387,7 +390,7 @@ private ScheduledFuture<?> scheduleStreamTimeout() {
                     logger.error(message);
                     rollback();
                     cancelStreamTimeout();
-                    responseObserver.onCompleted();
+                    responseObserver.onError(Status.DEADLINE_EXCEEDED.withDescription(message).asRuntimeException());
                 }, streamTimeout.plus(DEBOUNCE_INTERVAL).toNanos(), TimeUnit.NANOSECONDS);
             }
         };

src/main/resources/application.properties

@@ -1,8 +1,8 @@
 spring.datasource.url=${DB_URL:jdbc:informix-sqli://localhost:2021/informixoltp:INFORMIXSERVER=informixoltp_tcp}
 spring.datasource.username=${DB_USERNAME:informix}
 spring.datasource.password=${DB_PASSWORD:1nf0rm1x}
-spring.datasource.hikari.connectionTimeout=20000
+spring.datasource.hikari.connectionTimeout=30000
 spring.datasource.hikari.maximumPoolSize=20
-spring.datasource.hikari.leak-detection-threshold: 60000
+spring.datasource.hikari.leak-detection-threshold=90000
 grpc.server.port=9090
 logging.pattern.console=%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %X{AWS-XRAY-TRACE-ID} [%p] %M - %m%n