Automatically reconnect Kubernetes watcher when closed exceptionally (#…

…6023)

Motivation:

A watcher in `KubernetesEndpointGroup` automatically reconnects when it
fails to connect to the remote peer. However, it does not reconnect when
a `WatcherException` is raised.

```
io.fabric8.kubernetes.client.WatcherException: too old resource version: 573375490 (573377297)
	at io.fabric8.kubernetes.client.dsl.internal.AbstractWatchManager.onStatus(AbstractWatchManager.java:401)
	at io.fabric8.kubernetes.client.dsl.internal.AbstractWatchManager.onMessage(AbstractWatchManager.java:369)
	at io.fabric8.kubernetes.client.dsl.internal.WatcherWebSocketListener.onMessage(WatcherWebSocketListener.java:52)
	at com.linecorp.armeria.client.kubernetes.ArmeriaWebSocket.onNext(ArmeriaWebSocket.java:106)
	at com.linecorp.armeria.client.kubernetes.ArmeriaWebSocket.onNext(ArmeriaWebSocket.java:37)
	at com.linecorp.armeria.common.stream.DefaultStreamMessage.notifySubscriberWithElements(DefaultStreamMessage.java:412)
        ...
Caused by: io.fabric8.kubernetes.client.KubernetesClientException: too old resource version: 573375490 (573377297)
	... 62 common frames omitted
```

I don't know why `too old resource version` was raised but an important
thing is that watchers should not be stopped until
`KubernetesEndpointGroup` is closed.

Modifications:

- Refactor `KuberntesEndpointGroup` to start watchers asynchronously.
- Automatically restart `Watcher`s when `onClose(WatcherException)` is
invoked.
- Add more logs that I think might be useful.
  - Also make the log formats consistent
- Debounce the update of endpoints to prevent
`EndpointGroup.whenReady()` from completing with a small number of
endpoints.
- The purpose is to prevent a few endpoints from receiving too much
traffic when a watcher is newly created.

Result:

`KubernetesEndpointGroup` automatically reconnects a `Watcher` when
`WatcherException` is raised.

kubernetes/src/main/java/com/linecorp/armeria/client/kubernetes/ArmeriaHttpClientFactory.java

@@ -17,6 +17,8 @@
 package com.linecorp.armeria.client.kubernetes;
 
 import com.linecorp.armeria.client.WebClientBuilder;
+import com.linecorp.armeria.client.websocket.WebSocketClient;
+import com.linecorp.armeria.client.websocket.WebSocketClientBuilder;
 
 import io.fabric8.kubernetes.client.http.HttpClient;
 
@@ -36,4 +38,11 @@ public HttpClient.Builder newBuilder() {
     protected void additionalConfig(WebClientBuilder builder) {
         // no default implementation
     }
+
+    /**
+     * Subclasses may use this to apply additional configuration for {@link WebSocketClient}.
+     */
+    protected void additionalWebSocketConfig(WebSocketClientBuilder builder) {
+        // no default implementation
+    }
 }

kubernetes/src/main/java/com/linecorp/armeria/client/kubernetes/ArmeriaWebSocketClient.java

@@ -29,6 +29,7 @@
 
 import com.linecorp.armeria.client.RequestOptions;
 import com.linecorp.armeria.client.websocket.WebSocketClient;
+import com.linecorp.armeria.client.websocket.WebSocketClientBuilder;
 import com.linecorp.armeria.client.websocket.WebSocketClientHandshakeException;
 import com.linecorp.armeria.client.websocket.WebSocketSession;
 import com.linecorp.armeria.common.HttpHeaderNames;
@@ -69,10 +70,12 @@ private WebSocketClient webSocketClient() {
             if (webSocketClient0 != null) {
                 return webSocketClient0;
             }
-            webSocketClient0 = WebSocketClient.builder()
-                                              .factory(armeriaHttpClientBuilder.clientFactory(true))
-                                              .aggregateContinuation(true)
-                                              .build();
+            final WebSocketClientBuilder webSocketClientBuilder =
+                    WebSocketClient.builder()
+                                   .factory(armeriaHttpClientBuilder.clientFactory(true))
+                                   .aggregateContinuation(true);
+            armeriaHttpClientBuilder.getClientFactory().additionalWebSocketConfig(webSocketClientBuilder);
+            webSocketClient0 = webSocketClientBuilder.build();
             this.webSocketClient = webSocketClient0;
             return webSocketClient0;
         } finally {