[FLINK-4398] [query] Fix race in KvStateServerHandlerTest

Successful request are reported asynchronously after the Netty channel
write has succeeded. The test checked the number of successful requests
too early.

flink-runtime/src/main/java/org/apache/flink/runtime/query/netty/KvStateServerHandler.java

@@ -37,6 +37,7 @@
 import java.io.IOException;
 import java.util.Objects;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
 
 /**
  * This handler dispatches asynchronous tasks, which query {@link KvState}
@@ -286,7 +287,8 @@ private class QueryResultWriteListener implements ChannelFutureListener {
 
 			@Override
 			public void operationComplete(ChannelFuture future) throws Exception {
-				long durationMillis = (System.nanoTime() - creationNanos) / 1_000_000;
+				long durationNanos = System.nanoTime() - creationNanos;
+				long durationMillis = TimeUnit.MILLISECONDS.convert(durationNanos, TimeUnit.NANOSECONDS);
 
 				if (future.isSuccess()) {
 					stats.reportSuccessfulRequest(durationMillis);

flink-runtime/src/test/java/org/apache/flink/runtime/query/netty/KvStateServerHandlerTest.java

@@ -52,6 +52,7 @@
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import static org.junit.Assert.assertEquals;
@@ -150,6 +151,13 @@ public void testSimpleQuery() throws Exception {
 		assertEquals(expectedValue, actualValue);
 
 		assertEquals(stats.toString(), 1, stats.getNumRequests());
+
+		// Wait for async successful request report
+		long deadline = System.nanoTime() + TimeUnit.NANOSECONDS.convert(30, TimeUnit.SECONDS);
+		while (stats.getNumSuccessful() != 1 && System.nanoTime() <= deadline) {
+			Thread.sleep(10);
+		}
+
 		assertEquals(stats.toString(), 1, stats.getNumSuccessful());
 	}