Updates to Armeria 0.87 and addresses deprecations (#2631)

pom.xml

@@ -48,7 +48,7 @@
 
     <main.basedir>${project.basedir}</main.basedir>
 
-    <armeria.version>0.86.0</armeria.version>
+    <armeria.version>0.87.0</armeria.version>
     <!-- This is from armeria, but be careful to avoid >= v20 apis -->
     <guava.version>27.1-jre</guava.version>
 

zipkin-collector/scribe/src/main/java/zipkin2/collector/scribe/ScribeInboundHandler.java

@@ -162,7 +162,7 @@ void maybeReadPayload(ChannelHandlerContext ctx) {
           buf.release();
         }
       } else {
-        returned.writeBytes(content.array(), content.offset(), content.length());
+        returned.writeBytes(content.array());
       }
 
       if (responseIndex == previouslySentResponseIndex + 1) {

zipkin-server/src/main/java/zipkin2/server/internal/MetricsHealthController.java

@@ -96,6 +96,6 @@ public HttpResponse getHealth() throws JsonProcessingException {
     ResponseHeaders headers = ResponseHeaders.builder(statusMapper.mapStatus(health.getStatus()))
       .contentType(MediaType.JSON)
       .setInt(HttpHeaderNames.CONTENT_LENGTH, body.length).build();
-    return HttpResponse.of(headers, HttpData.of(body));
+    return HttpResponse.of(headers, HttpData.wrap(body));
   }
 }

zipkin-server/src/main/java/zipkin2/server/internal/RegisterZipkinHealthIndicators.java

@@ -23,8 +23,7 @@
 // This is an application listener to ensure the graph is fully constructed before doing health
 public final class RegisterZipkinHealthIndicators implements ApplicationListener {
 
-  @Override
-  public void onApplicationEvent(ApplicationEvent event) {
+  @Override public void onApplicationEvent(ApplicationEvent event) {
     if (!(event instanceof ApplicationReadyEvent)) return;
     ConfigurableListableBeanFactory beanFactory =
         ((ApplicationReadyEvent) event).getApplicationContext().getBeanFactory();

zipkin-server/src/main/java/zipkin2/server/internal/ZipkinHealthIndicator.java

@@ -40,8 +40,7 @@ static final class ComponentHealthIndicator implements HealthIndicator {
     }
 
     /** synchronized to prevent overlapping requests to a storage backend */
-    @Override
-    public synchronized Health health() {
+    @Override public synchronized Health health() {
       CheckResult result = component.check();
       return result.ok() ? Health.up().build() : Health.down((Exception) result.error()).build();
     }

zipkin-server/src/main/java/zipkin2/server/internal/ZipkinHttpCollector.java

@@ -14,7 +14,7 @@
 package zipkin2.server.internal;
 
 import com.linecorp.armeria.client.encoding.GzipStreamDecoderFactory;
-import com.linecorp.armeria.common.AggregatedHttpMessage;
+import com.linecorp.armeria.common.AggregatedHttpRequest;
 import com.linecorp.armeria.common.HttpData;
 import com.linecorp.armeria.common.HttpHeaderNames;
 import com.linecorp.armeria.common.HttpRequest;
@@ -115,15 +115,15 @@ HttpResponse validateAndStoreSpans(SpanBytesDecoder decoder, ServiceRequestConte
         return null;
       }
 
+      final HttpData content;
       try {
-        final HttpData content;
-        try {
-          content = UnzippingBytesRequestConverter.convertRequest(ctx, msg);
-        } catch (IllegalArgumentException e) {
-          result.onError(e);
-          return null;
-        }
+        content = UnzippingBytesRequestConverter.convertRequest(ctx, msg);
+      } catch (IllegalArgumentException e) {
+        result.onError(e);
+        return null;
+      }
 
+      try {
         // logging already handled upstream in UnzippingBytesRequestConverter where request context exists
         if (content.isEmpty()) {
           result.onSuccess(null);
@@ -134,9 +134,7 @@ HttpResponse validateAndStoreSpans(SpanBytesDecoder decoder, ServiceRequestConte
         if (content instanceof ByteBufHolder) {
           nioBuffer = ((ByteBufHolder) content).content().nioBuffer();
         } else {
-          // Currently this will happen for gzip spans. Need to fix armeria's gzip decoder to allow
-          // returning pooled buffers on request.
-          nioBuffer = ByteBuffer.wrap(content.array(), content.offset(), content.length());
+          nioBuffer = ByteBuffer.wrap(content.array());
         }
 
         try {
@@ -161,7 +159,7 @@ HttpResponse validateAndStoreSpans(SpanBytesDecoder decoder, ServiceRequestConte
         // UnzippingBytesRequestConverter handles incrementing message and bytes
         collector.accept(spans, result);
       } finally {
-        ReferenceCountUtil.release(msg.content());
+        ReferenceCountUtil.release(content);
       }
 
       return null;
@@ -170,7 +168,7 @@ HttpResponse validateAndStoreSpans(SpanBytesDecoder decoder, ServiceRequestConte
     return HttpResponse.from(result);
   }
 
-  static void maybeLog(String prefix, ServiceRequestContext ctx, AggregatedHttpMessage request) {
+  static void maybeLog(String prefix, ServiceRequestContext ctx, AggregatedHttpRequest request) {
     if (!LOGGER.isDebugEnabled()) return;
     LOGGER.debug("{} sent by clientAddress->{}, userAgent->{}",
       prefix, ctx.clientAddress(), request.headers().get(HttpHeaderNames.USER_AGENT)
@@ -205,7 +203,7 @@ final class CompletableCallback extends CompletableFuture<HttpResponse>
 final class UnzippingBytesRequestConverter {
   static final GzipStreamDecoderFactory GZIP_DECODER_FACTORY = new GzipStreamDecoderFactory();
 
-  static HttpData convertRequest(ServiceRequestContext ctx, AggregatedHttpMessage request) {
+  static HttpData convertRequest(ServiceRequestContext ctx, AggregatedHttpRequest request) {
     ZipkinHttpCollector.metrics.incrementMessages();
     String encoding = request.headers().get(HttpHeaderNames.CONTENT_ENCODING);
     HttpData content = request.content();
@@ -214,6 +212,7 @@ static HttpData convertRequest(ServiceRequestContext ctx, AggregatedHttpMessage
       // The implementation of the armeria decoder is to return an empty body on failure
       if (content.isEmpty()) {
         ZipkinHttpCollector.maybeLog("Malformed gzip body", ctx, request);
+        ReferenceCountUtil.release(content);
         throw new IllegalArgumentException("Cannot gunzip spans");
       }
     }

zipkin-server/src/main/java/zipkin2/server/internal/ZipkinQueryApiV2.java

@@ -13,7 +13,7 @@
  */
 package zipkin2.server.internal;
 
-import com.linecorp.armeria.common.AggregatedHttpMessage;
+import com.linecorp.armeria.common.AggregatedHttpResponse;
 import com.linecorp.armeria.common.HttpData;
 import com.linecorp.armeria.common.HttpHeaderNames;
 import com.linecorp.armeria.common.HttpStatus;
@@ -70,7 +70,7 @@ public class ZipkinQueryApiV2 {
   }
 
   @Get("/api/v2/dependencies")
-  public AggregatedHttpMessage getDependencies(
+  public AggregatedHttpResponse getDependencies(
     @Param("endTs") long endTs,
     @Param("lookback") Optional<Long> lookback) throws IOException {
     Call<List<DependencyLink>> call =
@@ -79,28 +79,28 @@ public AggregatedHttpMessage getDependencies(
   }
 
   @Get("/api/v2/services")
-  public AggregatedHttpMessage getServiceNames() throws IOException {
+  public AggregatedHttpResponse getServiceNames() throws IOException {
     List<String> serviceNames = storage.serviceAndSpanNames().getServiceNames().execute();
     serviceCount = serviceNames.size();
     return maybeCacheNames(serviceCount > 3, serviceNames);
   }
 
   @Get("/api/v2/spans")
-  public AggregatedHttpMessage getSpanNames(@Param("serviceName") String serviceName) throws IOException {
+  public AggregatedHttpResponse getSpanNames(@Param("serviceName") String serviceName) throws IOException {
     List<String> spanNames = storage.serviceAndSpanNames().getSpanNames(serviceName).execute();
     return maybeCacheNames(serviceCount > 3, spanNames);
   }
 
   @Get("/api/v2/remoteServices")
-  public AggregatedHttpMessage getRemoteServiceNames(@Param("serviceName") String serviceName)
+  public AggregatedHttpResponse getRemoteServiceNames(@Param("serviceName") String serviceName)
     throws IOException {
     List<String> remoteServiceNames =
       storage.serviceAndSpanNames().getRemoteServiceNames(serviceName).execute();
     return maybeCacheNames(serviceCount > 3, remoteServiceNames);
   }
 
   @Get("/api/v2/traces")
-  public AggregatedHttpMessage getTraces(
+  public AggregatedHttpResponse getTraces(
     @Param("serviceName") Optional<String> serviceName,
     @Param("remoteServiceName") Optional<String> remoteServiceName,
     @Param("spanName") Optional<String> spanName,
@@ -129,19 +129,19 @@ public AggregatedHttpMessage getTraces(
   }
 
   @Get("/api/v2/trace/{traceIdHex}")
-  public AggregatedHttpMessage getTrace(@Param("traceIdHex") String traceIdHex) throws IOException {
+  public AggregatedHttpResponse getTrace(@Param("traceIdHex") String traceIdHex) throws IOException {
     List<Span> trace = storage.spanStore().getTrace(traceIdHex).execute();
     if (trace == null) {
-      return AggregatedHttpMessage.of(HttpStatus.NOT_FOUND, MediaType.PLAIN_TEXT_UTF_8,
+      return AggregatedHttpResponse.of(HttpStatus.NOT_FOUND, MediaType.PLAIN_TEXT_UTF_8,
         traceIdHex + " not found");
     }
     return jsonResponse(SpanBytesEncoder.JSON_V2.encodeList(trace));
   }
 
-  static AggregatedHttpMessage jsonResponse(byte[] body) {
-    return AggregatedHttpMessage.of(ResponseHeaders.builder(200)
+  static AggregatedHttpResponse jsonResponse(byte[] body) {
+    return AggregatedHttpResponse.of(ResponseHeaders.builder(200)
       .contentType(MediaType.JSON)
-      .setInt(HttpHeaderNames.CONTENT_LENGTH, body.length).build(), HttpData.of(body));
+      .setInt(HttpHeaderNames.CONTENT_LENGTH, body.length).build(), HttpData.wrap(body));
   }
 
   static final WriteBuffer.Writer<String> QUOTED_STRING_WRITER = new WriteBuffer.Writer<String>() {
@@ -157,12 +157,12 @@ static AggregatedHttpMessage jsonResponse(byte[] body) {
   };
 
   @Get("/api/v2/autocompleteKeys")
-  public AggregatedHttpMessage getAutocompleteKeys() {
+  public AggregatedHttpResponse getAutocompleteKeys() {
     return maybeCacheNames(true, autocompleteKeys);
   }
 
   @Get("/api/v2/autocompleteValues")
-  public AggregatedHttpMessage getAutocompleteValues(@Param("key") String key) throws IOException {
+  public AggregatedHttpResponse getAutocompleteValues(@Param("key") String key) throws IOException {
     List<String> values = storage.autocompleteTags().getValues(key).execute();
     return maybeCacheNames(values.size() > 3, values);
   }
@@ -172,7 +172,7 @@ public AggregatedHttpMessage getAutocompleteValues(@Param("key") String key) thr
    * empty results, users have more questions. We assume caching becomes a concern when zipkin is in
    * active use, and active use usually implies more than 3 services.
    */
-  AggregatedHttpMessage maybeCacheNames(boolean shouldCacheControl, List<String> values) {
+  AggregatedHttpResponse maybeCacheNames(boolean shouldCacheControl, List<String> values) {
     Collections.sort(values);
     byte[] body = JsonCodec.writeList(QUOTED_STRING_WRITER, values);
     ResponseHeadersBuilder headers = ResponseHeaders.builder(200)
@@ -184,7 +184,7 @@ AggregatedHttpMessage maybeCacheNames(boolean shouldCacheControl, List<String> v
         CacheControl.maxAge(namesMaxAge, TimeUnit.SECONDS).mustRevalidate().getHeaderValue()
       );
     }
-    return AggregatedHttpMessage.of(headers.build(), HttpData.of(body));
+    return AggregatedHttpResponse.of(headers.build(), HttpData.wrap(body));
   }
 
   // This is inlined here as there isn't enough re-use to warrant it being in the zipkin2 library

zipkin-server/src/main/java/zipkin2/server/internal/ZipkinServerConfiguration.java

@@ -31,6 +31,7 @@
 import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.boot.actuate.autoconfigure.metrics.MeterRegistryCustomizer;
 import org.springframework.boot.actuate.health.HealthAggregator;
+import org.springframework.boot.actuate.health.HealthIndicatorRegistry;
 import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -44,13 +45,13 @@
 import org.springframework.core.type.AnnotatedTypeMetadata;
 import org.springframework.web.servlet.config.annotation.ViewControllerRegistry;
 import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
-import zipkin2.server.internal.throttle.ZipkinStorageThrottleProperties;
 import zipkin2.collector.CollectorMetrics;
 import zipkin2.collector.CollectorSampler;
 import zipkin2.server.internal.brave.TracingStorageComponent;
+import zipkin2.server.internal.throttle.ThrottledStorageComponent;
+import zipkin2.server.internal.throttle.ZipkinStorageThrottleProperties;
 import zipkin2.storage.InMemoryStorage;
 import zipkin2.storage.StorageComponent;
-import zipkin2.server.internal.throttle.ThrottledStorageComponent;
 
 @Configuration
 @ImportAutoConfiguration(ArmeriaSpringActuatorAutoConfiguration.class)
@@ -86,8 +87,7 @@ ZipkinHealthIndicator zipkinHealthIndicator(HealthAggregator healthAggregator) {
     return new ZipkinHealthIndicator(healthAggregator);
   }
 
-  @Override
-  public void addViewControllers(ViewControllerRegistry registry) {
+  @Override public void addViewControllers(ViewControllerRegistry registry) {
     registry.addRedirectViewController("/info", "/actuator/info");
   }
 

zipkin-server/src/main/java/zipkin2/server/internal/brave/TracingStorageComponent.java

@@ -73,13 +73,11 @@ public Call<List<Span>> getTrace(String traceId) {
       return new TracedCall<>(tracer, delegate.getTrace(traceId), "get-trace");
     }
 
-    @Override
-    public Call<List<String>> getServiceNames() {
+    @Override @Deprecated public Call<List<String>> getServiceNames() {
       return new TracedCall<>(tracer, delegate.getServiceNames(), "get-service-names");
     }
 
-    @Override
-    public Call<List<String>> getSpanNames(String serviceName) {
+    @Override @Deprecated public Call<List<String>> getSpanNames(String serviceName) {
       return new TracedCall<>(tracer, delegate.getSpanNames(serviceName), "get-span-names");
     }
 

zipkin-server/src/main/java/zipkin2/server/internal/prometheus/ZipkinPrometheusMetricsConfiguration.java

@@ -20,7 +20,7 @@
 import com.linecorp.armeria.common.Response;
 import com.linecorp.armeria.common.logging.RequestLog;
 import com.linecorp.armeria.common.logging.RequestLogAvailability;
-import com.linecorp.armeria.server.PathMapping;
+import com.linecorp.armeria.server.Route;
 import com.linecorp.armeria.server.Service;
 import com.linecorp.armeria.server.ServiceRequestContext;
 import com.linecorp.armeria.server.SimpleDecoratingService;
@@ -73,7 +73,8 @@ public class ZipkinPrometheusMetricsConfiguration {
   @Order(1)
   ArmeriaServerConfigurator notFoundMetricCollector() {
     // Use glob instead of catch-all to avoid adding it to the trie router.
-    return sb -> sb.service(PathMapping.ofGlob("/**"), (ctx, req) -> HttpResponse.of(HttpStatus.NOT_FOUND));
+    return sb -> sb.service(Route.builder().glob("/**").build(),
+      (ctx, req) -> HttpResponse.of(HttpStatus.NOT_FOUND));
   }
 
   static final class MetricCollectingService<I extends Request, O extends Response>

zipkin-server/src/main/java/zipkin2/server/internal/ui/ZipkinUiConfiguration.java

@@ -21,11 +21,11 @@
 import com.linecorp.armeria.common.HttpResponse;
 import com.linecorp.armeria.common.HttpStatus;
 import com.linecorp.armeria.common.MediaType;
+import com.linecorp.armeria.common.ServerCacheControl;
+import com.linecorp.armeria.common.ServerCacheControlBuilder;
 import com.linecorp.armeria.server.AbstractHttpService;
 import com.linecorp.armeria.server.HttpService;
 import com.linecorp.armeria.server.RedirectService;
-import com.linecorp.armeria.server.ServerCacheControl;
-import com.linecorp.armeria.server.ServerCacheControlBuilder;
 import com.linecorp.armeria.server.Service;
 import com.linecorp.armeria.server.ServiceRequestContext;
 import com.linecorp.armeria.server.encoding.HttpEncodingService;
@@ -126,8 +126,8 @@ String processedIndexHtml(Resource indexHtml) {
       legacyIndex = HttpFileBuilder.ofResource("zipkin-ui/index.html");
       lensIndex = HttpFileBuilder.ofResource("zipkin-lens/index.html");
     } else {
-      legacyIndex = HttpFileBuilder.of(HttpData.of(processedIndexHtml().getBytes(UTF_8)));
-      lensIndex = HttpFileBuilder.of(HttpData.of(processedLensIndexHtml().getBytes(UTF_8)));
+      legacyIndex = HttpFileBuilder.of(HttpData.wrap(processedIndexHtml().getBytes(UTF_8)));
+      lensIndex = HttpFileBuilder.of(HttpData.wrap(processedLensIndexHtml().getBytes(UTF_8)));
     }
 
     ServerCacheControl maxAgeMinute = new ServerCacheControlBuilder().maxAgeSeconds(60).build();
@@ -153,7 +153,7 @@ String processedIndexHtml(Resource indexHtml) {
 
     byte[] config = new ObjectMapper().writeValueAsBytes(ui);
     return sb -> {
-      sb.service("/zipkin/config.json", HttpFileBuilder.of(HttpData.of(config))
+      sb.service("/zipkin/config.json", HttpFileBuilder.of(HttpData.wrap(config))
         .cacheControl(new ServerCacheControlBuilder().maxAgeSeconds(600).build())
         .contentType(MediaType.JSON_UTF_8)
         .build()

zipkin-server/src/test/kotlin/zipkin2/server/internal/ui/ZipkinUiConfigurationTest.kt

@@ -13,7 +13,7 @@
  */
 package zipkin2.server.internal.ui
 
-import com.linecorp.armeria.common.AggregatedHttpMessage
+import com.linecorp.armeria.common.AggregatedHttpResponse
 import com.linecorp.armeria.common.HttpHeaderNames
 import com.linecorp.armeria.common.HttpMethod
 import com.linecorp.armeria.common.HttpRequest
@@ -160,7 +160,7 @@ class ZipkinUiConfigurationTest {
       .contains("<base href=\"/\">")
   }
 
-  private fun serveIndex(vararg cookies: Cookie): AggregatedHttpMessage {
+  private fun serveIndex(vararg cookies: Cookie): AggregatedHttpResponse {
     val headers = RequestHeaders.builder(HttpMethod.GET, "/")
     val encodedCookies = ClientCookieEncoder.LAX.encode(*cookies)
     if (encodedCookies != null) {