Remove inefficient namespace clone. (#21552)

Testing shows this is causing ~ 5MB/s of throughput on the platform. This is not needed since we can simply modify the already present Json node instead of a cloned object. This should help both CPU and GC pressure.
airbytehq · Jan 18, 2023 · da34c78 · da34c78
1 parent e092a4e
commit da34c78
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 15 deletions.
diff --git a/airbyte-commons-worker/src/main/java/io/airbyte/workers/internal/NamespacingMapper.java b/airbyte-commons-worker/src/main/java/io/airbyte/workers/internal/NamespacingMapper.java
@@ -59,9 +59,8 @@ public ConfiguredAirbyteCatalog mapCatalog(final ConfiguredAirbyteCatalog inputC
   }
 
   @Override
-  public AirbyteMessage mapMessage(final AirbyteMessage inputMessage) {
-    if (inputMessage.getType() == Type.RECORD) {
-      final AirbyteMessage message = Jsons.clone(inputMessage);
+  public AirbyteMessage mapMessage(final AirbyteMessage message) {
+    if (message.getType() == Type.RECORD) {
       // Default behavior if namespaceDefinition is not set is to follow SOURCE
       if (namespaceDefinition != null) {
         if (namespaceDefinition.equals(NamespaceDefinitionType.DESTINATION)) {
@@ -73,7 +72,7 @@ public AirbyteMessage mapMessage(final AirbyteMessage inputMessage) {
       message.getRecord().setStream(transformStreamName(message.getRecord().getStream(), streamPrefix));
       return message;
     }
-    return inputMessage;
+    return message;
   }
 
   private static String formatNamespace(final String sourceNamespace, final String namespaceFormat) {

diff --git a/airbyte-commons-worker/src/test/java/io/airbyte/workers/internal/NamespacingMapperTest.java b/airbyte-commons-worker/src/test/java/io/airbyte/workers/internal/NamespacingMapperTest.java
@@ -14,6 +14,7 @@
 import io.airbyte.protocol.models.Field;
 import io.airbyte.protocol.models.JsonSchemaType;
 import io.airbyte.workers.test_utils.AirbyteMessageUtils;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 class NamespacingMapperTest {
@@ -28,14 +29,19 @@ class NamespacingMapperTest {
       STREAM_NAME,
       INPUT_NAMESPACE,
       Field.of(FIELD_NAME, JsonSchemaType.STRING));
-  private static final AirbyteMessage RECORD_MESSAGE = createRecordMessage();
+  private AirbyteMessage RECORD_MESSAGE;
 
   private static AirbyteMessage createRecordMessage() {
     final AirbyteMessage message = AirbyteMessageUtils.createRecordMessage(STREAM_NAME, FIELD_NAME, BLUE);
     message.getRecord().withNamespace(INPUT_NAMESPACE);
     return message;
   }
 
+  @BeforeEach
+  void setUp() {
+    RECORD_MESSAGE = createRecordMessage();
+  }
+
   @Test
   void testSourceNamespace() {
     final NamespacingMapper mapper = new NamespacingMapper(NamespaceDefinitionType.SOURCE, null, OUTPUT_PREFIX);
@@ -51,11 +57,12 @@ void testSourceNamespace() {
     assertEquals(expectedCatalog, actualCatalog);
 
     final AirbyteMessage originalMessage = Jsons.clone(RECORD_MESSAGE);
+    assertEquals(originalMessage, RECORD_MESSAGE);
+
     final AirbyteMessage expectedMessage = AirbyteMessageUtils.createRecordMessage(OUTPUT_PREFIX + STREAM_NAME, FIELD_NAME, BLUE);
     expectedMessage.getRecord().withNamespace(INPUT_NAMESPACE);
-    final AirbyteMessage actualMessage = mapper.mapMessage(RECORD_MESSAGE);
 
-    assertEquals(originalMessage, RECORD_MESSAGE);
+    final AirbyteMessage actualMessage = mapper.mapMessage(RECORD_MESSAGE);
     assertEquals(expectedMessage, actualMessage);
   }
 
@@ -100,10 +107,10 @@ void testDestinationNamespace() {
     assertEquals(expectedCatalog, actualCatalog);
 
     final AirbyteMessage originalMessage = Jsons.clone(RECORD_MESSAGE);
+    assertEquals(originalMessage, RECORD_MESSAGE);
+
     final AirbyteMessage expectedMessage = AirbyteMessageUtils.createRecordMessage(OUTPUT_PREFIX + STREAM_NAME, FIELD_NAME, BLUE);
     final AirbyteMessage actualMessage = mapper.mapMessage(RECORD_MESSAGE);
-
-    assertEquals(originalMessage, RECORD_MESSAGE);
     assertEquals(expectedMessage, actualMessage);
   }
 
@@ -122,11 +129,11 @@ void testCustomFormatWithVariableNamespace() {
     assertEquals(expectedCatalog, actualCatalog);
 
     final AirbyteMessage originalMessage = Jsons.clone(RECORD_MESSAGE);
+    assertEquals(originalMessage, RECORD_MESSAGE);
+
     final AirbyteMessage expectedMessage = AirbyteMessageUtils.createRecordMessage(OUTPUT_PREFIX + STREAM_NAME, FIELD_NAME, BLUE);
     expectedMessage.getRecord().withNamespace(expectedNamespace);
     final AirbyteMessage actualMessage = mapper.mapMessage(RECORD_MESSAGE);
-
-    assertEquals(originalMessage, RECORD_MESSAGE);
     assertEquals(expectedMessage, actualMessage);
   }
 
@@ -145,11 +152,11 @@ void testCustomFormatWithoutVariableNamespace() {
     assertEquals(expectedCatalog, actualCatalog);
 
     final AirbyteMessage originalMessage = Jsons.clone(RECORD_MESSAGE);
+    assertEquals(originalMessage, RECORD_MESSAGE);
+
     final AirbyteMessage expectedMessage = AirbyteMessageUtils.createRecordMessage(OUTPUT_PREFIX + STREAM_NAME, FIELD_NAME, BLUE);
     expectedMessage.getRecord().withNamespace(expectedNamespace);
     final AirbyteMessage actualMessage = mapper.mapMessage(RECORD_MESSAGE);
-
-    assertEquals(originalMessage, RECORD_MESSAGE);
     assertEquals(expectedMessage, actualMessage);
   }
 
@@ -194,13 +201,13 @@ void testEmptyPrefix() {
     assertEquals(expectedCatalog, actualCatalog);
 
     final AirbyteMessage originalMessage = Jsons.clone(RECORD_MESSAGE);
+    assertEquals(originalMessage, RECORD_MESSAGE);
+
     final AirbyteMessage expectedMessage = AirbyteMessageUtils.createRecordMessage(
         STREAM_NAME,
         FIELD_NAME, BLUE);
     expectedMessage.getRecord().withNamespace(INPUT_NAMESPACE);
     final AirbyteMessage actualMessage = mapper.mapMessage(RECORD_MESSAGE);
-
-    assertEquals(originalMessage, RECORD_MESSAGE);
     assertEquals(expectedMessage, actualMessage);
   }