Time out connections on the IOThread level

Time out connections on the IOThread level instead of leaving this to Apache.
Keep old connections alive for a while after timeout and keep polling them
such that, if the old connection hits a different real behind a VIP than the
new connection we'll still get the replies.

4 files changed, 34 insertions, 28 deletions

diff --git a/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/QueueBoundsTest.java b/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/QueueBoundsTest.java
index 1f875e0dd72..8ff2566ead1 100644
--- a/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/QueueBoundsTest.java
+++ b/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/QueueBoundsTest.java
@@ -24,6 +24,7 @@ import static com.yahoo.vespa.http.client.TestUtils.writeDocument;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.core.IsEqual.equalTo;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.fail;
 
@@ -249,12 +250,12 @@ public class QueueBoundsTest {
                 }
                 int errors = 0;
                 for (Result result : session.results()) {
-                    assertThat(result.getDetails().size(), is(1));
+                    assertEquals(1, result.getDetails().size());
                     if (! result.isSuccess()) {
                         errors++;
                     }
                 }
-                assertThat(errors, is(1));
+                assertEquals(1, errors);
             } finally {
                 feeder.stop();
             }
diff --git a/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/V3HttpAPITest.java b/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/V3HttpAPITest.java
index 1d70ce953e4..6907e24009b 100644
--- a/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/V3HttpAPITest.java
+++ b/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/V3HttpAPITest.java
@@ -21,7 +21,11 @@ import static com.yahoo.vespa.http.client.TestUtils.getResults;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.CoreMatchers.nullValue;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 
 /**
  *
@@ -79,16 +83,15 @@ public class V3HttpAPITest {
 
             writeDocument(session);
             Map<String, Result> results = getResults(session, 1);
-            assertThat(results.size(), is(1));
+            assertEquals(1, results.size());
 
             TestDocument document = documents.get(0);
             Result r = results.remove(document.getDocumentId());
-            assertThat(r, not(nullValue()));
-            if (conditionNotMet) {
-                assertThat(r.getDetails().iterator().next().getResultType(), is(Result.ResultType.CONDITION_NOT_MET));
-            }
-            assertThat(r.getDetails().toString(), r.isSuccess(), is(false));
-            assertThat(results.isEmpty(), is(true));
+            assertNotNull(r);
+            if (conditionNotMet)
+                assertEquals(Result.ResultType.CONDITION_NOT_MET, r.getDetails().iterator().next().getResultType());
+            assertFalse(r.getDetails().toString(), r.isSuccess());
+            assertTrue(results.isEmpty());
         }
     }
 
@@ -99,14 +102,14 @@ public class V3HttpAPITest {
 
             writeDocuments(session);
             Map<String, Result> results = getResults(session, documents.size());
-            assertThat(results.size(), is(documents.size()));
+            assertEquals(documents.size(), results.size());
 
             for (TestDocument document : documents) {
                 Result r = results.remove(document.getDocumentId());
                 assertThat(r, not(nullValue()));
                 assertThat(r.getDetails().toString(), r.isSuccess(), is(true));
             }
-            assertThat(results.isEmpty(), is(true));
+            assertTrue(results.isEmpty());
         }
     }
 
@@ -169,15 +172,15 @@ public class V3HttpAPITest {
             writeDocuments(session);
 
             Map<String, Result> results = getResults(session, documents.size());
-            assertThat(results.size(), is(documents.size()));
+            assertEquals(documents.size(), results.size());
 
             for (TestDocument document : documents) {
                 Result r = results.remove(document.getDocumentId());
-                assertThat(r, not(nullValue()));
-                assertThat(r.getDetails().toString(), r.isSuccess(), is(false));
-                assertThat(r.getDetails().iterator().next().getResultType(), is(Result.ResultType.TRANSITIVE_ERROR));
+                assertNotNull(r);
+                assertFalse(r.getDetails().toString(), r.isSuccess());
+                assertEquals(Result.ResultType.TRANSITIVE_ERROR, r.getDetails().iterator().next().getResultType());
             }
-            assertThat(results.isEmpty(), is(true));
+            assertTrue(results.isEmpty());
         }
     }
 
diff --git a/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/core/communication/ApacheGatewayConnectionTest.java b/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/core/communication/ApacheGatewayConnectionTest.java
index 494f901d8d7..4169e7e0ecf 100644
--- a/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/core/communication/ApacheGatewayConnectionTest.java
+++ b/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/core/communication/ApacheGatewayConnectionTest.java
@@ -76,7 +76,7 @@ public class ApacheGatewayConnectionTest {
         apacheGatewayConnection.handshake();
         documents.add(createDoc(docId, vespaDocContent, true));
 
-        apacheGatewayConnection.writeOperations(documents);
+        apacheGatewayConnection.write(documents);
     }
 
     @Test(expected=IllegalArgumentException.class)
@@ -100,7 +100,7 @@ public class ApacheGatewayConnectionTest {
                         "clientId");
         apacheGatewayConnection.connect();
         final List<Document> documents = new ArrayList<>();
-        apacheGatewayConnection.writeOperations(documents);
+        apacheGatewayConnection.write(documents);
     }
 
     @Test
@@ -145,7 +145,7 @@ public class ApacheGatewayConnectionTest {
 
         documents.add(createDoc(docId, vespaDocContent, true));
 
-        apacheGatewayConnection.writeOperations(documents);
+        apacheGatewayConnection.write(documents);
     }
 
     @Test
@@ -210,7 +210,7 @@ public class ApacheGatewayConnectionTest {
 
         documents.add(doc);
 
-        apacheGatewayConnection.writeOperations(documents);
+        apacheGatewayConnection.write(documents);
     }
 
     @Test
@@ -248,8 +248,8 @@ public class ApacheGatewayConnectionTest {
 
         List<Document> documents = new ArrayList<>();
         documents.add(createDoc("42", "content", true));
-        apacheGatewayConnection.writeOperations(documents);
-        apacheGatewayConnection.writeOperations(documents);
+        apacheGatewayConnection.write(documents);
+        apacheGatewayConnection.write(documents);
 
         verify(headerProvider, times(3)).getHeaderValue(); // 1x connect(), 2x writeOperations()
     }
@@ -274,7 +274,7 @@ public class ApacheGatewayConnectionTest {
         apacheGatewayConnection.connect();
         apacheGatewayConnection.handshake();
 
-        apacheGatewayConnection.writeOperations(Collections.singletonList(createDoc("42", "content", true)));
+        apacheGatewayConnection.write(Collections.singletonList(createDoc("42", "content", true)));
     }
 
     private static ApacheGatewayConnection.HttpClientFactory mockHttpClientFactory(HttpExecuteMock httpExecuteMock) throws IOException {
diff --git a/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/core/communication/IOThreadTest.java b/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/core/communication/IOThreadTest.java
index b313daa12cd..75df8a78b86 100644
--- a/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/core/communication/IOThreadTest.java
+++ b/vespa-http-client/src/test/java/com/yahoo/vespa/http/client/core/communication/IOThreadTest.java
@@ -16,6 +16,7 @@ import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
@@ -90,9 +91,10 @@ public class IOThreadTest {
                             0,
                             0,
                             maxInFlightRequests,
-                            localQueueTimeOut,
+                            Duration.ofMillis(localQueueTimeOut),
                             documentQueue,
                             0,
+                            Duration.ofSeconds(15),
                             10);
     }
 
@@ -101,7 +103,7 @@ public class IOThreadTest {
         when(apacheGatewayConnection.connect()).thenReturn(true);
         InputStream serverResponse = new ByteArrayInputStream(
                 (docId1 + " OK Doc{20}fed").getBytes(StandardCharsets.UTF_8));
-        when(apacheGatewayConnection.writeOperations(any())).thenReturn(serverResponse);
+        when(apacheGatewayConnection.write(any())).thenReturn(serverResponse);
         setupEndpointResultQueueMock( "nope", docId1, true, exceptionMessage);
         try (IOThread ioThread = createIOThread(10000, 10000)) {
             ioThread.post(doc1);
@@ -112,7 +114,7 @@ public class IOThreadTest {
     @Test
     public void requireThatSingleDocumentWriteErrorIsHandledProperly() throws Exception {
         when(apacheGatewayConnection.connect()).thenReturn(true);
-        when(apacheGatewayConnection.writeOperations(any())).thenThrow(new IOException(exceptionMessage));
+        when(apacheGatewayConnection.write(any())).thenThrow(new IOException(exceptionMessage));
         setupEndpointResultQueueMock(doc1.getOperationId(), "nope", true, exceptionMessage);
         try (IOThread ioThread = createIOThread(10000, 10000)) {
             ioThread.post(doc1);
@@ -125,7 +127,7 @@ public class IOThreadTest {
         when(apacheGatewayConnection.connect()).thenReturn(true);
         InputStream serverResponse = new ByteArrayInputStream(
                 (docId2 + " OK Doc{20}fed").getBytes(StandardCharsets.UTF_8));
-        when(apacheGatewayConnection.writeOperations(any()))
+        when(apacheGatewayConnection.write(any()))
                 .thenThrow(new IOException(exceptionMessage))
                 .thenReturn(serverResponse);
         latch = new CountDownLatch(2);
@@ -143,7 +145,7 @@ public class IOThreadTest {
         when(apacheGatewayConnection.connect()).thenReturn(false);
         InputStream serverResponse = new ByteArrayInputStream(
                 ("").getBytes(StandardCharsets.UTF_8));
-        when(apacheGatewayConnection.writeOperations(any()))
+        when(apacheGatewayConnection.write(any()))
                 .thenReturn(serverResponse);
         setupEndpointResultQueueMock(doc1.getOperationId(), "nope", true,
                 "java.lang.Exception: Not sending document operation, timed out in queue after");