Emit the state to remove in the airbyte empty source

airbyte-config/config-models/src/main/java/io/airbyte/config/helpers/StateMessageHelper.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.config.helpers;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.config.StateType;
+import io.airbyte.config.StateWrapper;
+import io.airbyte.protocol.models.AirbyteStateMessage;
+import io.airbyte.protocol.models.AirbyteStateMessage.AirbyteStateType;
+import java.util.List;
+import java.util.Optional;
+
+public class StateMessageHelper {
+
+  public static class AirbyteStateMessageListTypeReference extends TypeReference<List<AirbyteStateMessage>> {}
+
+  /**
+   * This a takes a json blob state and tries return either a legacy state in the format of a json
+   * object or a state message with the new format which is a list of airbyte state message.
+   *
+   * @param state
+   * @return Either a json blob (on the left) or a structure state message.
+   */
+  public static Optional<StateWrapper> getTypedState(JsonNode state) {
+    if (state == null) {
+      return Optional.empty();
+    } else {
+      try {
+        List<AirbyteStateMessage> stateMessages = Jsons.object(state, new AirbyteStateMessageListTypeReference());
+        if (stateMessages.size() == 1 && stateMessages.get(0).getStateType() == AirbyteStateType.GLOBAL) {
+          return Optional.of(new StateWrapper()
+              .withStateType(StateType.GLOBAL)
+              .withGlobal(stateMessages.get(0)));
+        } else if (stateMessages.size() >= 1
+            && stateMessages.stream().filter(stateMessage -> stateMessage.getStateType() != AirbyteStateType.STREAM).toList().isEmpty()) {
+          return Optional.of(new StateWrapper()
+              .withStateType(StateType.STREAM)
+              .withStateMessages(stateMessages));
+        } else {
+          throw new IllegalStateException("Unexpected state blob");
+        }
+
+      } catch (final IllegalArgumentException e) {
+        return Optional.of(new StateWrapper()
+            .withStateType(StateType.LEGACY)
+            .withLegacyState(state));
+      }
+    }
+    /*
+     * if (state == null) { // return Either.right(new ArrayList<>()); } try { return
+     * Either.right(Jsons.object(state, new AirbyteStateMessageListTypeReference())); } catch (final
+     * IllegalArgumentException e) { return Either.left(state); }
+     */
+  }
+
+}

airbyte-config/config-models/src/main/resources/types/StateType.yaml

@@ -0,0 +1,10 @@
+---
+"$schema": http://json-schema.org/draft-07/schema#
+"$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-config/models/src/main/resources/types/StateType.yaml
+title: StateType
+description: State Types
+type: string
+enum:
+  - global
+  - stream
+  - legacy

airbyte-config/config-models/src/main/resources/types/StateWrapper.yaml

@@ -0,0 +1,26 @@
+---
+"$schema": http://json-schema.org/draft-07/schema#
+"$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-config/models/src/main/resources/types/StateWrapper.yaml
+title: StateWrapper
+description: Wrapper around the different type of States
+type: object
+additionalProperties: false
+required:
+  - stateType
+properties:
+  stateType:
+    description: The type of the state being wrapped
+    "$ref": StateType.yaml
+  legacyState:
+    description: Legacy State for states that haven't been migrated yet
+    type: object
+    existingJavaType: com.fasterxml.jackson.databind.JsonNode
+  global:
+    description: Representation of the shared
+    type: object
+    existingJavaType: io.airbyte.protocol.models.AirbyteStateMessage
+  stateMessages:
+    type: array
+    items:
+      type: object
+      existingJavaType: io.airbyte.protocol.models.AirbyteStateMessage

airbyte-config/config-models/src/test/java/io/airbyte/config/helpers/StateMessageHelperTest.java

@@ -0,0 +1,113 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.config.helpers;
+
+import com.google.common.collect.Lists;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.config.StateType;
+import io.airbyte.config.StateWrapper;
+import io.airbyte.protocol.models.AirbyteGlobalState;
+import io.airbyte.protocol.models.AirbyteStateMessage;
+import io.airbyte.protocol.models.AirbyteStateMessage.AirbyteStateType;
+import io.airbyte.protocol.models.AirbyteStreamState;
+import io.airbyte.protocol.models.StreamDescriptor;
+import java.util.Optional;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class StateMessageHelperTest {
+
+  @Test
+  public void testEmpty() {
+    Optional<StateWrapper> stateWrapper = StateMessageHelper.getTypedState(null);
+    Assertions.assertThat(stateWrapper).isEmpty();
+  }
+
+  @Test
+  public void testLegacy() {
+    Optional<StateWrapper> stateWrapper = StateMessageHelper.getTypedState(Jsons.emptyObject());
+    Assertions.assertThat(stateWrapper).isNotEmpty();
+    Assertions.assertThat(stateWrapper.get().getStateType()).isEqualTo(StateType.LEGACY);
+  }
+
+  @Test
+  public void testGlobal() {
+    AirbyteStateMessage stateMessage = new AirbyteStateMessage()
+        .withStateType(AirbyteStateType.GLOBAL)
+        .withGlobal(
+            new AirbyteGlobalState()
+                .withSharedState(Jsons.emptyObject())
+                .withStreamStates(Lists.newArrayList(
+                    new AirbyteStreamState().withStreamDescriptor(new StreamDescriptor().withName("a")).withStreamState(Jsons.emptyObject()),
+                    new AirbyteStreamState().withStreamDescriptor(new StreamDescriptor().withName("b")).withStreamState(Jsons.emptyObject()))));
+    Optional<StateWrapper> stateWrapper = StateMessageHelper.getTypedState(Jsons.jsonNode(Lists.newArrayList(stateMessage)));
+    Assertions.assertThat(stateWrapper).isNotEmpty();
+    Assertions.assertThat(stateWrapper.get().getStateType()).isEqualTo(StateType.GLOBAL);
+    Assertions.assertThat(stateWrapper.get().getGlobal()).isEqualTo(stateMessage);
+  }
+
+  @Test
+  public void testStream() {
+    AirbyteStateMessage stateMessage1 = new AirbyteStateMessage()
+        .withStateType(AirbyteStateType.STREAM)
+        .withStream(
+            new AirbyteStreamState().withStreamDescriptor(new StreamDescriptor().withName("a")).withStreamState(Jsons.emptyObject()));
+    AirbyteStateMessage stateMessage2 = new AirbyteStateMessage()
+        .withStateType(AirbyteStateType.STREAM)
+        .withStream(
+            new AirbyteStreamState().withStreamDescriptor(new StreamDescriptor().withName("b")).withStreamState(Jsons.emptyObject()));
+    Optional<StateWrapper> stateWrapper = StateMessageHelper.getTypedState(Jsons.jsonNode(Lists.newArrayList(stateMessage1, stateMessage2)));
+    Assertions.assertThat(stateWrapper).isNotEmpty();
+    Assertions.assertThat(stateWrapper.get().getStateType()).isEqualTo(StateType.STREAM);
+    Assertions.assertThat(stateWrapper.get().getStateMessages()).containsExactlyInAnyOrder(stateMessage1, stateMessage2);
+  }
+
+  @Test
+  public void testInvalidMixedState() {
+    AirbyteStateMessage stateMessage1 = new AirbyteStateMessage()
+        .withStateType(AirbyteStateType.STREAM)
+        .withStream(
+            new AirbyteStreamState().withStreamDescriptor(new StreamDescriptor().withName("a")).withStreamState(Jsons.emptyObject()));
+    AirbyteStateMessage stateMessage2 = new AirbyteStateMessage()
+        .withStateType(AirbyteStateType.GLOBAL)
+        .withGlobal(
+            new AirbyteGlobalState()
+                .withSharedState(Jsons.emptyObject())
+                .withStreamStates(Lists.newArrayList(
+                    new AirbyteStreamState().withStreamDescriptor(new StreamDescriptor().withName("a")).withStreamState(Jsons.emptyObject()),
+                    new AirbyteStreamState().withStreamDescriptor(new StreamDescriptor().withName("b")).withStreamState(Jsons.emptyObject()))));
+    Assertions.assertThatThrownBy(() -> StateMessageHelper.getTypedState(Jsons.jsonNode(Lists.newArrayList(stateMessage1, stateMessage2))))
+        .isInstanceOf(IllegalStateException.class);
+  }
+
+  @Test
+  public void testDuplicatedGlobalState() {
+    AirbyteStateMessage stateMessage1 = new AirbyteStateMessage()
+        .withStateType(AirbyteStateType.GLOBAL)
+        .withGlobal(
+            new AirbyteGlobalState()
+                .withSharedState(Jsons.emptyObject())
+                .withStreamStates(Lists.newArrayList(
+                    new AirbyteStreamState().withStreamDescriptor(new StreamDescriptor().withName("a")).withStreamState(Jsons.emptyObject()),
+                    new AirbyteStreamState().withStreamDescriptor(new StreamDescriptor().withName("b")).withStreamState(Jsons.emptyObject()))));
+    AirbyteStateMessage stateMessage2 = new AirbyteStateMessage()
+        .withStateType(AirbyteStateType.GLOBAL)
+        .withGlobal(
+            new AirbyteGlobalState()
+                .withSharedState(Jsons.emptyObject())
+                .withStreamStates(Lists.newArrayList(
+                    new AirbyteStreamState().withStreamDescriptor(new StreamDescriptor().withName("a")).withStreamState(Jsons.emptyObject()),
+                    new AirbyteStreamState().withStreamDescriptor(new StreamDescriptor().withName("b")).withStreamState(Jsons.emptyObject()))));
+    Assertions.assertThatThrownBy(() -> StateMessageHelper.getTypedState(Jsons.jsonNode(Lists.newArrayList(stateMessage1, stateMessage2))))
+        .isInstanceOf(IllegalStateException.class);
+  }
+
+  @Test
+  public void testEmptyStateList() {
+    Assertions.assertThatThrownBy(() -> StateMessageHelper.getTypedState(Jsons.jsonNode(Lists.newArrayList())))
+        .isInstanceOf(IllegalStateException.class);
+  }
+
+}