Merge branch 'main' into gc-connection-metrics

.github/workflows/actions_build.yml

@@ -32,7 +32,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        on: [ ubicloud-standard-8, macos-12, windows-latest ]
+        on: [ ubicloud-standard-8, macos-latest, windows-latest ]
         java: [ 21 ]
         include:
           - java: 8

.github/workflows/publish-site.yml

@@ -1,6 +1,12 @@
 name: Publish Armeria site
 
 on:
+  workflow_dispatch:
+    inputs:
+      version:
+        description: 'Release Version'
+        required: true
+        type: string
   push:
     tags:
       - armeria-*
@@ -32,7 +38,11 @@ jobs:
 
       - name: Build the site
         run: |
-          ./gradlew --no-daemon --stacktrace  --max-workers=2 --parallel -PgithubToken=${{ secrets.GITHUB_TOKEN }} site
+          if [ "${{ github.event_name }}" == "workflow_dispatch" ]; then
+            ./gradlew --no-daemon --stacktrace --max-workers=2 --parallel -PgithubToken=${{ secrets.GITHUB_TOKEN }} -Pversion=${{ inputs.version }} site
+          else
+            ./gradlew --no-daemon --stacktrace --max-workers=2 --parallel -PgithubToken=${{ secrets.GITHUB_TOKEN }} site
+          fi
 
       - name: Deploy the site
         uses: peaceiris/actions-gh-pages@v4

build.gradle

@@ -489,3 +489,7 @@ allprojects {
         }
     }
 }
+
+configure(projectsWithFlags('java', 'publish')) {
+    failOnVersionConflict(libs.protobuf.java)
+}

core/src/main/java/com/linecorp/armeria/client/AbstractHttpResponseDecoder.java

@@ -62,8 +62,7 @@ public HttpResponseWrapper addResponse(@Nullable AbstractHttpRequestHandler requ
                                            int id, DecodedHttpResponse res,
                                            ClientRequestContext ctx, EventLoop eventLoop) {
         final HttpResponseWrapper newRes =
-                new HttpResponseWrapper(requestHandler, res, eventLoop, ctx,
-                                        ctx.responseTimeoutMillis(), ctx.maxResponseLength());
+                new HttpResponseWrapper(requestHandler, res, eventLoop, ctx, ctx.maxResponseLength());
         final HttpResponseWrapper oldRes = responses.put(id, newRes);
         keepAliveHandler().increaseNumRequests();
 

core/src/main/java/com/linecorp/armeria/client/HttpResponseWrapper.java

@@ -58,23 +58,21 @@ class HttpResponseWrapper implements StreamWriter<HttpObject> {
     private final EventLoop eventLoop;
     private final ClientRequestContext ctx;
     private final long maxContentLength;
-    private final long responseTimeoutMillis;
 
     private boolean responseStarted;
     private long contentLengthHeaderValue = -1;
 
     private boolean done;
     private boolean closed;
 
-    HttpResponseWrapper(@Nullable AbstractHttpRequestHandler requestHandler,
-                        DecodedHttpResponse delegate, EventLoop eventLoop, ClientRequestContext ctx,
-                        long responseTimeoutMillis, long maxContentLength) {
+    HttpResponseWrapper(@Nullable AbstractHttpRequestHandler requestHandler, DecodedHttpResponse delegate,
+                        EventLoop eventLoop, ClientRequestContext ctx, long maxContentLength) {
+
         this.requestHandler = requestHandler;
         this.delegate = delegate;
         this.eventLoop = eventLoop;
         this.ctx = ctx;
         this.maxContentLength = maxContentLength;
-        this.responseTimeoutMillis = responseTimeoutMillis;
     }
 
     void handle100Continue(ResponseHeaders responseHeaders) {
@@ -327,7 +325,6 @@ public String toString() {
                                    .add("eventLoop", eventLoop)
                                    .add("responseStarted", responseStarted)
                                    .add("maxContentLength", maxContentLength)
-                                   .add("responseTimeoutMillis", responseTimeoutMillis)
                                    .add("contentLengthHeaderValue", contentLengthHeaderValue)
                                    .add("delegate", delegate)
                                    .toString();

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ClientChannelHandler.java

@@ -100,8 +100,7 @@ public HttpResponseWrapper addResponse(@Nullable AbstractHttpRequestHandler requ
                                            int id, DecodedHttpResponse decodedHttpResponse,
                                            ClientRequestContext ctx, EventLoop eventLoop) {
         assert res == null;
-        res = new WebSocketHttp1ResponseWrapper(decodedHttpResponse, eventLoop, ctx,
-                                                ctx.responseTimeoutMillis(), ctx.maxResponseLength());
+        res = new WebSocketHttp1ResponseWrapper(decodedHttpResponse, eventLoop, ctx, ctx.maxResponseLength());
         return res;
     }
 

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ResponseWrapper.java

@@ -26,9 +26,8 @@
 final class WebSocketHttp1ResponseWrapper extends HttpResponseWrapper {
 
     WebSocketHttp1ResponseWrapper(DecodedHttpResponse delegate,
-                                  EventLoop eventLoop, ClientRequestContext ctx,
-                                  long responseTimeoutMillis, long maxContentLength) {
-        super(null, delegate, eventLoop, ctx, responseTimeoutMillis, maxContentLength);
+                                  EventLoop eventLoop, ClientRequestContext ctx, long maxContentLength) {
+        super(null, delegate, eventLoop, ctx, maxContentLength);
         WebSocketClientUtil.setClosingResponseTask(ctx, cause -> {
             super.close(cause, false);
         });

core/src/main/java/com/linecorp/armeria/client/endpoint/DynamicEndpointGroup.java

@@ -17,6 +17,7 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableList.toImmutableList;
+import static com.linecorp.armeria.internal.client.endpoint.EndpointToStringUtil.toShortString;
 import static com.linecorp.armeria.internal.common.util.CollectionUtil.truncate;
 import static java.util.Objects.requireNonNull;
 
@@ -32,6 +33,9 @@
 import java.util.concurrent.locks.Lock;
 import java.util.function.Consumer;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
@@ -51,6 +55,8 @@
  */
 public class DynamicEndpointGroup extends AbstractEndpointGroup implements ListenableAsyncCloseable {
 
+    private static final Logger logger = LoggerFactory.getLogger(DynamicEndpointGroup.class);
+
     /**
      * Returns a newly created builder.
      */
@@ -223,6 +229,8 @@ protected final void addEndpoint(Endpoint e) {
             final List<Endpoint> newEndpointsUnsorted = Lists.newArrayList(endpoints);
             newEndpointsUnsorted.add(e);
             endpoints = newEndpoints = ImmutableList.sortedCopyOf(newEndpointsUnsorted);
+            logger.info("An endpoint has been added: {}. Current endpoints: {}",
+                        toShortString(e), toShortString(newEndpoints));
         } finally {
             endpointsLock.unlock();
         }
@@ -238,12 +246,17 @@ protected final void removeEndpoint(Endpoint e) {
         final List<Endpoint> newEndpoints;
         endpointsLock.lock();
         try {
-            if (!allowEmptyEndpoints && endpoints.size() == 1) {
+            final List<Endpoint> oldEndpoints = endpoints;
+            if (!allowEmptyEndpoints && oldEndpoints.size() == 1) {
                 return;
             }
-            endpoints = newEndpoints = endpoints.stream()
-                                                .filter(endpoint -> !endpoint.equals(e))
-                                                .collect(toImmutableList());
+            endpoints = newEndpoints = oldEndpoints.stream()
+                                                   .filter(endpoint -> !endpoint.equals(e))
+                                                   .collect(toImmutableList());
+            if (endpoints.size() != oldEndpoints.size()) {
+                logger.info("An endpoint has been removed: {}. Current endpoints: {}",
+                            toShortString(e), toShortString(newEndpoints));
+            }
         } finally {
             endpointsLock.unlock();
         }
@@ -266,6 +279,7 @@ protected final void setEndpoints(Iterable<Endpoint> endpoints) {
                 return;
             }
             this.endpoints = newEndpoints;
+            logger.info("New endpoints have been set: {}", toShortString(newEndpoints));
         } finally {
             endpointsLock.unlock();
         }
@@ -376,7 +390,7 @@ public String toString() {
     protected final String toString(Consumer<? super StringBuilder> builderMutator) {
         final StringBuilder buf = new StringBuilder();
         buf.append(getClass().getSimpleName());
-        buf.append("{selectionStrategy=").append(selectionStrategy.getClass());
+        buf.append("{selector=").append(toStringSelector());
         buf.append(", allowsEmptyEndpoints=").append(allowEmptyEndpoints);
         buf.append(", initialized=").append(initialEndpointsFuture.isDone());
         buf.append(", numEndpoints=").append(endpoints.size());
@@ -385,6 +399,21 @@ protected final String toString(Consumer<? super StringBuilder> builderMutator)
         return buf.append('}').toString();
     }
 
+    /**
+     * Returns the string representation of the {@link EndpointSelector} of this {@link DynamicEndpointGroup}.
+     * If the {@link EndpointSelector} is not created yet, it returns the class name of the
+     * {@link EndpointSelectionStrategy}.
+     */
+    protected String toStringSelector() {
+        final EndpointSelector endpointSelector = selector.get();
+        if (endpointSelector == null) {
+            // Return selection strategy if selector is not created yet.
+            return selectionStrategy.getClass().toString();
+        }
+
+        return endpointSelector.toString();
+    }
+
     private class InitialEndpointsFuture extends EventLoopCheckingFuture<List<Endpoint>> {
 
         @Override

core/src/main/java/com/linecorp/armeria/client/endpoint/RoundRobinStrategy.java

@@ -19,6 +19,8 @@
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import com.google.common.base.MoreObjects;
+
 import com.linecorp.armeria.client.ClientRequestContext;
 import com.linecorp.armeria.client.Endpoint;
 import com.linecorp.armeria.common.annotation.Nullable;
@@ -57,5 +59,12 @@ public Endpoint selectNow(ClientRequestContext ctx) {
             final int currentSequence = sequence.getAndIncrement();
             return endpoints.get(Math.abs(currentSequence % endpoints.size()));
         }
+
+        @Override
+        public String toString() {
+            return MoreObjects.toStringHelper(this)
+                              .add("endpoints", group().endpoints())
+                              .toString();
+        }
     }
 }

core/src/main/java/com/linecorp/armeria/client/endpoint/StickyEndpointSelectionStrategy.java

@@ -20,6 +20,7 @@
 import java.util.List;
 import java.util.function.ToLongFunction;
 
+import com.google.common.base.MoreObjects;
 import com.google.common.hash.Hashing;
 
 import com.linecorp.armeria.client.ClientRequestContext;
@@ -84,7 +85,6 @@ private static final class StickyEndpointSelector extends AbstractEndpointSelect
         @Nullable
         @Override
         public Endpoint selectNow(ClientRequestContext ctx) {
-
             final List<Endpoint> endpoints = group().endpoints();
             if (endpoints.isEmpty()) {
                 return null;
@@ -94,5 +94,12 @@ public Endpoint selectNow(ClientRequestContext ctx) {
             final int nearest = Hashing.consistentHash(key, endpoints.size());
             return endpoints.get(nearest);
         }
+
+        @Override
+        public String toString() {
+            return MoreObjects.toStringHelper(this)
+                              .add("endpoints", group().endpoints())
+                              .toString();
+        }
     }
 }

core/src/main/java/com/linecorp/armeria/client/endpoint/WeightRampingUpStrategy.java

@@ -22,11 +22,10 @@
 import static com.linecorp.armeria.client.endpoint.WeightRampingUpStrategyBuilder.defaultTransition;
 import static com.linecorp.armeria.internal.client.endpoint.EndpointAttributeKeys.createdAtNanos;
 import static com.linecorp.armeria.internal.client.endpoint.EndpointAttributeKeys.hasCreatedAtNanos;
+import static com.linecorp.armeria.internal.client.endpoint.EndpointToStringUtil.toShortString;
 import static java.util.Objects.requireNonNull;
 
-import java.util.ArrayDeque;
 import java.util.ArrayList;
-import java.util.Deque;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -37,6 +36,9 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableList;
@@ -76,6 +78,8 @@
  */
 final class WeightRampingUpStrategy implements EndpointSelectionStrategy {
 
+    private static final Logger logger = LoggerFactory.getLogger(WeightRampingUpStrategy.class);
+
     private static final Ticker defaultTicker = Ticker.systemTicker();
     private static final WeightedRandomDistributionEndpointSelector EMPTY_SELECTOR =
             new WeightedRandomDistributionEndpointSelector(ImmutableList.of());
@@ -130,8 +134,6 @@ final class RampingUpEndpointWeightSelector extends AbstractEndpointSelector {
 
         private final List<Endpoint> endpointsFinishedRampingUp = new ArrayList<>();
 
-        @VisibleForTesting
-        final Deque<EndpointsRampingUpEntry> endpointsRampingUp = new ArrayDeque<>();
         @VisibleForTesting
         final Map<Long, EndpointsRampingUpEntry> rampingUpWindowsMap = new HashMap<>();
         private Object2LongOpenHashMap<Endpoint> endpointCreatedTimestamps = new Object2LongOpenHashMap<>();
@@ -233,7 +235,25 @@ private void buildEndpointSelector() {
                             endpointAndStep.endpoint().withWeight(endpointAndStep.currentWeight()));
                 }
             }
-            endpointSelector = new WeightedRandomDistributionEndpointSelector(targetEndpointsBuilder.build());
+            final List<Endpoint> endpoints = targetEndpointsBuilder.build();
+            if (rampingUpWindowsMap.isEmpty()) {
+                logger.info("Finished ramping up. endpoints: {}", toShortString(endpoints));
+            } else {
+                logger.debug("Ramping up. endpoints: {}", toShortString(endpoints));
+            }
+
+            boolean found = false;
+            for (Endpoint endpoint : endpoints) {
+                if (endpoint.weight() > 0) {
+                    found = true;
+                    break;
+                }
+            }
+            if (!found) {
+                logger.warn("No valid endpoint with weight > 0. endpoints: {}", toShortString(endpoints));
+            }
+
+            endpointSelector = new WeightedRandomDistributionEndpointSelector(endpoints);
         }
 
         @VisibleForTesting
@@ -288,6 +308,15 @@ private void close() {
                 lock.unlock();
             }
         }
+
+        @Override
+        public String toString() {
+            return MoreObjects.toStringHelper(this)
+                              .add("endpointSelector", endpointSelector)
+                              .add("endpointsFinishedRampingUp", endpointsFinishedRampingUp)
+                              .add("rampingUpWindowsMap", rampingUpWindowsMap)
+                              .toString();
+        }
     }
 
     private static int numStep(long rampingUpIntervalNanos, Ticker ticker, long createTimestamp) {

core/src/main/java/com/linecorp/armeria/client/endpoint/WeightRampingUpStrategyBuilder.java

@@ -43,9 +43,17 @@ public final class WeightRampingUpStrategyBuilder {
     static final int DEFAULT_TOTAL_STEPS = 10;
     static final int DEFAULT_RAMPING_UP_TASK_WINDOW_MILLIS = 500;
     static final EndpointWeightTransition DEFAULT_LINEAR_TRANSITION =
-            (endpoint, currentStep, totalSteps) ->
-                    // currentStep is never greater than totalSteps so we can cast long to int.
-                    Ints.saturatedCast((long) endpoint.weight() * currentStep / totalSteps);
+            (endpoint, currentStep, totalSteps) -> {
+                // currentStep is never greater than totalSteps so we can cast long to int.
+                final int currentWeight =
+                        Ints.saturatedCast((long) endpoint.weight() * currentStep / totalSteps);
+                if (endpoint.weight() > 0 && currentWeight == 0) {
+                    // If the original weight is not 0,
+                    // we should return 1 to make sure the endpoint is selected.
+                    return 1;
+                }
+                return currentWeight;
+            };
     static final EndpointWeightTransition defaultTransition = EndpointWeightTransition.linear();
 
     private EndpointWeightTransition transition = defaultTransition;