Replacing InboundMessage with NativeInboundMessage for deprecation (o…

…pensearch-project#13126)

* Replacing InboundMessage with NativeInboundMessage for deprecation

Signed-off-by: Vacha Shah <vachshah@amazon.com>

* Removing InboundMessage class

Signed-off-by: Vacha Shah <vachshah@amazon.com>

---------

Signed-off-by: Vacha Shah <vachshah@amazon.com>

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/transport/InboundAggregatorTests.java

@@ -42,6 +42,7 @@
 import org.opensearch.core.common.breaker.CircuitBreakingException;
 import org.opensearch.core.common.bytes.BytesArray;
 import org.opensearch.test.OpenSearchTestCase;
+import org.opensearch.transport.nativeprotocol.NativeInboundMessage;
 import org.junit.Before;
 
 import java.io.IOException;
@@ -107,7 +108,7 @@ public void testInboundAggregation() throws IOException {
  }
 
  // Signal EOS
- InboundMessage aggregated = aggregator.finishAggregation();
+ NativeInboundMessage aggregated = aggregator.finishAggregation();
 
  assertThat(aggregated, notNullValue());
  assertFalse(aggregated.isPing());
@@ -138,7 +139,7 @@ public void testInboundUnknownAction() throws IOException {
  assertEquals(0, content.refCount());
 
  // Signal EOS
- InboundMessage aggregated = aggregator.finishAggregation();
+ NativeInboundMessage aggregated = aggregator.finishAggregation();
 
  assertThat(aggregated, notNullValue());
  assertTrue(aggregated.isShortCircuit());
@@ -161,7 +162,7 @@ public void testCircuitBreak() throws IOException {
  content1.close();
 
  // Signal EOS
- InboundMessage aggregated1 = aggregator.finishAggregation();
+ NativeInboundMessage aggregated1 = aggregator.finishAggregation();
 
  assertEquals(0, content1.refCount());
  assertThat(aggregated1, notNullValue());
@@ -180,7 +181,7 @@ public void testCircuitBreak() throws IOException {
  content2.close();
 
  // Signal EOS
- InboundMessage aggregated2 = aggregator.finishAggregation();
+ NativeInboundMessage aggregated2 = aggregator.finishAggregation();
 
  assertEquals(1, content2.refCount());
  assertThat(aggregated2, notNullValue());
@@ -199,7 +200,7 @@ public void testCircuitBreak() throws IOException {
  content3.close();
 
  // Signal EOS
- InboundMessage aggregated3 = aggregator.finishAggregation();
+ NativeInboundMessage aggregated3 = aggregator.finishAggregation();
 
  assertEquals(1, content3.refCount());
  assertThat(aggregated3, notNullValue());
@@ -263,7 +264,7 @@ public void testFinishAggregationWillFinishHeader() throws IOException {
  content.close();
 
  // Signal EOS
- InboundMessage aggregated = aggregator.finishAggregation();
+ NativeInboundMessage aggregated = aggregator.finishAggregation();
 
  assertThat(aggregated, notNullValue());
  assertFalse(header.needsToReadVariableHeader());