Merge branch 'main' into oss-12371

.github/workflows/detect-breaking-change.yml

@@ -1,6 +1,8 @@
 name: "Detect Breaking Changes"
 on:
-  pull_request
+  pull_request:
+    branches-ignore:
+      - main # This branch represents a to-be-released version of OpenSearch where breaking changes are allowed
 
 jobs:
   detect-breaking-change:

CHANGELOG.md

@@ -60,6 +60,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Enabled mockTelemetryPlugin for IT and fixed OOM issues ([#13054](https://github.com/opensearch-project/OpenSearch/pull/13054))
 - Fix implement mark() and markSupported() in class FilterStreamInput ([#13098](https://github.com/opensearch-project/OpenSearch/pull/13098))
 - Fix snapshot _status API to return correct status for partial snapshots ([#12812](https://github.com/opensearch-project/OpenSearch/pull/12812))
+- Ignore BaseRestHandler unconsumed content check as it's always consumed. ([#13290](https://github.com/opensearch-project/OpenSearch/pull/13290))
 
 ### Security
 

release-notes/opensearch.release-notes-1.3.16.md

@@ -0,0 +1,4 @@
+## 2024-04-18 Version 1.3.16 Release Notes
+
+### Upgrades
+- Bump `netty` from 4.1.107.Final to 4.1.109.Final ([#12924](https://github.com/opensearch-project/OpenSearch/pull/12924), [#13233](https://github.com/opensearch-project/OpenSearch/pull/13233))

rest-api-spec/src/main/resources/rest-api-spec/test/indices.forcemerge/10_basic.yml

@@ -31,8 +31,8 @@
 ---
 "Test primary_only parameter":
   - skip:
-      version: " - 2.99.99"
-      reason: "primary_only is available in 3.0+"
+      version: " - 2.12.99"
+      reason: "primary_only is available in 2.13.0+"
 
   - do:
       indices.create:

server/src/main/java/org/opensearch/rest/BaseRestHandler.java

@@ -122,10 +122,6 @@ public final void handleRequest(RestRequest request, RestChannel channel, NodeCl
             throw new IllegalArgumentException(unrecognized(request, unconsumedParams, candidateParams, "parameter"));
         }
 
-        if (request.hasContent() && request.isContentConsumed() == false) {
-            throw new IllegalArgumentException("request [" + request.method() + " " + request.path() + "] does not support having a body");
-        }
-
         usageCount.increment();
         // execute the action
         action.accept(channel);

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

@@ -40,6 +40,7 @@
 import org.opensearch.core.common.bytes.BytesArray;
 import org.opensearch.core.common.bytes.BytesReference;
 import org.opensearch.core.common.bytes.CompositeBytesReference;
+import org.opensearch.transport.nativeprotocol.NativeInboundMessage;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -113,7 +114,7 @@ public void aggregate(ReleasableBytesReference content) {
         }
     }
 
-    public InboundMessage finishAggregation() throws IOException {
+    public NativeInboundMessage finishAggregation() throws IOException {
         ensureOpen();
         final ReleasableBytesReference releasableContent;
         if (isFirstContent()) {
@@ -127,7 +128,7 @@ public InboundMessage finishAggregation() throws IOException {
         }
 
         final BreakerControl breakerControl = new BreakerControl(circuitBreaker);
-        final InboundMessage aggregated = new InboundMessage(currentHeader, releasableContent, breakerControl);
+        final NativeInboundMessage aggregated = new NativeInboundMessage(currentHeader, releasableContent, breakerControl);
         boolean success = false;
         try {
             if (aggregated.getHeader().needsToReadVariableHeader()) {
@@ -142,7 +143,7 @@ public InboundMessage finishAggregation() throws IOException {
             if (isShortCircuited()) {
                 aggregated.close();
                 success = true;
-                return new InboundMessage(aggregated.getHeader(), aggregationException);
+                return new NativeInboundMessage(aggregated.getHeader(), aggregationException);
             } else {
                 success = true;
                 return aggregated;

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

@@ -51,6 +51,7 @@
 import org.opensearch.telemetry.tracing.Tracer;
 import org.opensearch.telemetry.tracing.channels.TraceableTcpTransportChannel;
 import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.transport.nativeprotocol.NativeInboundMessage;
 
 import java.io.EOFException;
 import java.io.IOException;
@@ -111,7 +112,7 @@ public void messageReceived(
         long slowLogThresholdMs,
         TransportMessageListener messageListener
     ) throws IOException {
-        InboundMessage inboundMessage = (InboundMessage) message;
+        NativeInboundMessage inboundMessage = (NativeInboundMessage) message;
         TransportLogger.logInboundMessage(channel, inboundMessage);
         if (inboundMessage.isPing()) {
             keepAlive.receiveKeepAlive(channel);
@@ -122,7 +123,7 @@ public void messageReceived(
 
     private void handleMessage(
         TcpChannel channel,
-        InboundMessage message,
+        NativeInboundMessage message,
         long startTime,
         long slowLogThresholdMs,
         TransportMessageListener messageListener
@@ -194,7 +195,7 @@ private Map<String, Collection<String>> extractHeaders(Map<String, String> heade
     private <T extends TransportRequest> void handleRequest(
         TcpChannel channel,
         Header header,
-        InboundMessage message,
+        NativeInboundMessage message,
         TransportMessageListener messageListener
     ) throws IOException {
         final String action = header.getActionName();

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

@@ -761,18 +761,6 @@ protected void serverAcceptedChannel(TcpChannel channel) {
      */
     protected abstract void stopInternal();
 
-    /**
-     * @deprecated use {@link #inboundMessage(TcpChannel, ProtocolInboundMessage)}
-     * Handles inbound message that has been decoded.
-     *
-     * @param channel the channel the message is from
-     * @param message the message
-     */
-    @Deprecated(since = "2.14.0", forRemoval = true)
-    public void inboundMessage(TcpChannel channel, InboundMessage message) {
-        inboundMessage(channel, (ProtocolInboundMessage) message);
-    }
-
     /**
      * Handles inbound message that has been decoded.
      *

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

@@ -40,6 +40,7 @@
 import org.opensearch.core.common.io.stream.InputStreamStreamInput;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.compress.CompressorRegistry;
+import org.opensearch.transport.nativeprotocol.NativeInboundMessage;
 
 import java.io.IOException;
 
@@ -64,7 +65,7 @@ static void logInboundMessage(TcpChannel channel, BytesReference message) {
         }
     }
 
-    static void logInboundMessage(TcpChannel channel, InboundMessage message) {
+    static void logInboundMessage(TcpChannel channel, NativeInboundMessage message) {
         if (logger.isTraceEnabled()) {
             try {
                 String logMessage = format(channel, message, "READ");
@@ -136,7 +137,7 @@ private static String format(TcpChannel channel, BytesReference message, String
         return sb.toString();
     }
 
-    private static String format(TcpChannel channel, InboundMessage message, String event) throws IOException {
+    private static String format(TcpChannel channel, NativeInboundMessage message, String event) throws IOException {
         final StringBuilder sb = new StringBuilder();
         sb.append(channel);
 

server/src/main/java/org/opensearch/transport/nativeprotocol/NativeInboundBytesHandler.java

@@ -16,7 +16,6 @@
 import org.opensearch.transport.InboundAggregator;
 import org.opensearch.transport.InboundBytesHandler;
 import org.opensearch.transport.InboundDecoder;
-import org.opensearch.transport.InboundMessage;
 import org.opensearch.transport.ProtocolInboundMessage;
 import org.opensearch.transport.StatsTracker;
 import org.opensearch.transport.TcpChannel;
@@ -32,7 +31,7 @@
 public class NativeInboundBytesHandler implements InboundBytesHandler {
 
     private static final ThreadLocal<ArrayList<Object>> fragmentList = ThreadLocal.withInitial(ArrayList::new);
-    private static final InboundMessage PING_MESSAGE = new InboundMessage(null, true);
+    private static final NativeInboundMessage PING_MESSAGE = new NativeInboundMessage(null, true);
 
     private final ArrayDeque<ReleasableBytesReference> pending;
     private final InboundDecoder decoder;
@@ -152,7 +151,7 @@ private void forwardFragments(
                 messageHandler.accept(channel, PING_MESSAGE);
             } else if (fragment == InboundDecoder.END_CONTENT) {
                 assert aggregator.isAggregating();
-                try (InboundMessage aggregated = aggregator.finishAggregation()) {
+                try (NativeInboundMessage aggregated = aggregator.finishAggregation()) {
                     statsTracker.markMessageReceived();
                     messageHandler.accept(channel, aggregated);
                 }

server/src/test/java/org/opensearch/action/admin/indices/forcemerge/RestForceMergeActionTests.java

@@ -32,45 +32,23 @@
 
 package org.opensearch.action.admin.indices.forcemerge;
 
-import org.opensearch.client.node.NodeClient;
-import org.opensearch.common.xcontent.json.JsonXContent;
-import org.opensearch.core.common.bytes.BytesArray;
-import org.opensearch.core.xcontent.MediaTypeRegistry;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.rest.RestRequest;
 import org.opensearch.rest.action.admin.indices.RestForceMergeAction;
-import org.opensearch.test.rest.FakeRestChannel;
 import org.opensearch.test.rest.FakeRestRequest;
 import org.opensearch.test.rest.RestActionTestCase;
 import org.junit.Before;
 
 import java.util.HashMap;
 import java.util.Map;
 
-import static org.hamcrest.Matchers.equalTo;
-import static org.mockito.Mockito.mock;
-
 public class RestForceMergeActionTests extends RestActionTestCase {
 
     @Before
     public void setUpAction() {
         controller().registerHandler(new RestForceMergeAction());
     }
 
-    public void testBodyRejection() throws Exception {
-        final RestForceMergeAction handler = new RestForceMergeAction();
-        String json = JsonXContent.contentBuilder().startObject().field("max_num_segments", 1).endObject().toString();
-        final FakeRestRequest request = new FakeRestRequest.Builder(NamedXContentRegistry.EMPTY).withContent(
-            new BytesArray(json),
-            MediaTypeRegistry.JSON
-        ).withPath("/_forcemerge").build();
-        IllegalArgumentException e = expectThrows(
-            IllegalArgumentException.class,
-            () -> handler.handleRequest(request, new FakeRestChannel(request, randomBoolean(), 1), mock(NodeClient.class))
-        );
-        assertThat(e.getMessage(), equalTo("request [GET /_forcemerge] does not support having a body"));
-    }
-
     public void testDeprecationMessage() {
         final Map<String, String> params = new HashMap<>();
         params.put("only_expunge_deletes", Boolean.TRUE.toString());

server/src/test/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeActionTests.java

@@ -91,8 +91,6 @@
 import static org.opensearch.node.remotestore.RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING;
 import static org.opensearch.test.ClusterServiceUtils.createClusterService;
 import static org.opensearch.test.ClusterServiceUtils.setState;
-import static org.opensearch.test.VersionUtils.randomCompatibleVersion;
-import static org.opensearch.test.VersionUtils.randomOpenSearchVersion;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
@@ -826,10 +824,9 @@ public void testDontAllowSwitchingCompatibilityModeForClusterWithMultipleVersion
         request.persistentSettings(intendedCompatibilityModeSettings);
 
         // two different but compatible open search versions for the discovery nodes
-        final Version version1 = randomOpenSearchVersion(random());
-        final Version version2 = randomCompatibleVersion(random(), version1);
+        final Version version1 = Version.V_2_13_0;
+        final Version version2 = Version.V_2_13_1;
 
-        assert version1.equals(version2) == false : "current nodes in the cluster must be of different versions";
         DiscoveryNode discoveryNode1 = new DiscoveryNode(
             UUIDs.base64UUID(),
             buildNewFakeTransportAddress(),