Merge branch 'opensearch-project:main' into indexing-tracing-instrume…

…ntation

libs/telemetry/src/main/java/org/opensearch/telemetry/tracing/DefaultTracer.java

@@ -25,7 +25,10 @@
  */
 @InternalApi
 class DefaultTracer implements Tracer {
-    static final String THREAD_NAME = "th_name";
+    /**
+     * Current thread name.
+     */
+    static final String THREAD_NAME = "thread.name";
 
     private final TracingTelemetry tracingTelemetry;
     private final TracerContextStorage<String, Span> tracerContextStorage;

server/src/main/java/org/opensearch/http/AbstractHttpServerTransport.java

@@ -298,6 +298,7 @@ static int resolvePublishPort(Settings settings, List<TransportAddress> boundAdd
     }
 
     public void onException(HttpChannel channel, Exception e) {
+        channel.handleException(e);
         if (lifecycle.started() == false) {
             // just close and ignore - we are already stopped and just need to make sure we release all resources
             CloseableChannel.closeChannel(channel);

server/src/main/java/org/opensearch/http/HttpChannel.java

@@ -43,6 +43,11 @@
  * @opensearch.internal
  */
 public interface HttpChannel extends CloseableChannel {
+    /**
+     * Notify HTTP channel that exception happens and the response may not be sent (for example, timeout)
+     * @param ex the exception being raised
+     */
+    default void handleException(Exception ex) {}
 
     /**
      * Sends an http response to the channel. The listener will be executed once the send process has been

server/src/main/java/org/opensearch/index/merge/MergeStats.java

@@ -82,7 +82,7 @@ public MergeStats(StreamInput in) throws IOException {
         totalStoppedTimeInMillis = in.readVLong();
         totalThrottledTimeInMillis = in.readVLong();
         totalBytesPerSecAutoThrottle = in.readVLong();
-        if (in.getVersion().onOrAfter(Version.V_3_0_0)) {
+        if (in.getVersion().onOrAfter(Version.V_2_11_0)) {
             unreferencedFileCleanUpsPerformed = in.readOptionalVLong();
         }
     }
@@ -297,7 +297,7 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeVLong(totalStoppedTimeInMillis);
         out.writeVLong(totalThrottledTimeInMillis);
         out.writeVLong(totalBytesPerSecAutoThrottle);
-        if (out.getVersion().onOrAfter(Version.V_3_0_0)) {
+        if (out.getVersion().onOrAfter(Version.V_2_11_0)) {
             out.writeOptionalVLong(unreferencedFileCleanUpsPerformed);
         }
     }

server/src/main/java/org/opensearch/telemetry/tracing/channels/TraceableHttpChannel.java

@@ -56,6 +56,13 @@ public static HttpChannel create(HttpChannel delegate, Span span, Tracer tracer)
         }
     }
 
+    @Override
+    public void handleException(Exception ex) {
+        span.addEvent("The HttpChannel was closed without sending the response");
+        span.setError(ex);
+        span.endSpan();
+    }
+
     @Override
     public void close() {
         delegate.close();

server/src/main/java/org/opensearch/telemetry/tracing/handler/TraceableTransportResponseHandler.java

@@ -101,6 +101,7 @@ public void handleRejection(Exception exp) {
         try (SpanScope scope = tracer.withSpanInScope(span)) {
             delegate.handleRejection(exp);
         } finally {
+            span.setError(exp);
             span.endSpan();
         }
     }

server/src/main/java/org/opensearch/transport/TransportResponseHandler.java

@@ -57,7 +57,7 @@ public interface TransportResponseHandler<T extends TransportResponse> extends W
      * It should be used to clear up the resources held by the {@link TransportResponseHandler}.
      * @param exp exception
      */
-    default void handleRejection(Exception exp) {};
+    default void handleRejection(Exception exp) {}
 
     default <Q extends TransportResponse> TransportResponseHandler<Q> wrap(Function<Q, T> converter, Writeable.Reader<Q> reader) {
         final TransportResponseHandler<T> self = this;