Filter invalid Singer Messages in the worker (#236)

Author: cgardens

dataline-singer/build.gradle

@@ -7,6 +7,11 @@ dependencies {
 }
 
 jsonSchema2Pojo {
+    // todo (cgardens) - this path cannot be the same as the one in dataline-configs:model,
+    //  otherwise they clobber each other, when we are accessing them as java resources.
+    //  we need to find a more graceful way of handling this issue, but for now this keeps
+    //  us moving forward.
+    source = files("${sourceSets.main.output.resourcesDir}/singer_json")
     targetDirectory = new File(project.buildDir, 'generated/src/gen/java/')
 
     targetPackage = 'io.dataline.singer'

dataline-singer/src/main/java/io/dataline/singer/SingerConfigSchema.java

@@ -0,0 +1,81 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Dataline
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.dataline.singer;
+
+import io.dataline.commons.io.IOs;
+import io.dataline.commons.resources.MoreResources;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.stream.Collectors;
+
+// todo (cgardens) - dedupe this with ConfigSchema.
+public enum SingerConfigSchema {
+
+  SINGER_MESSAGE("SingerMessage.json");
+
+  static final Path KNOWN_SCHEMAS_ROOT = prepareSchemas();
+  private static final String RESOURCE_DIR = "singer_json";
+
+  /*
+   * JsonReferenceProcessor relies on all of the json in consumes being in a file system (not in a
+   * jar). This method copies all of the json configs out of the jar into a temporary directory so
+   * that JsonReferenceProcessor can find them.
+   */
+  @SuppressWarnings("UnstableApiUsage")
+  private static Path prepareSchemas() {
+    try {
+      final List<String> filenames = MoreResources.listResources(SingerConfigSchema.class, RESOURCE_DIR)
+          .map(p -> p.getFileName().toString())
+          .filter(p -> p.endsWith(".json"))
+          .collect(Collectors.toList());
+
+      final Path configRoot = Files.createTempDirectory("schemas");
+      for (String filename : filenames) {
+        IOs.writeFile(
+            configRoot,
+            filename,
+            MoreResources.readResource(String.format("%s/%s", RESOURCE_DIR, filename)));
+      }
+
+      return configRoot;
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private final String schemaFilename;
+
+  SingerConfigSchema(final String schemaFilename) {
+    this.schemaFilename = schemaFilename;
+  }
+
+  public File getFile() {
+    return KNOWN_SCHEMAS_ROOT.resolve(schemaFilename).toFile();
+  }
+
+}

dataline-singer/src/main/resources/json/SingerCatalog.json renamed to dataline-singer/src/main/resources/singer_json/SingerCatalog.json

@@ -0,0 +1,49 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Dataline
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.dataline.singer;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import org.junit.jupiter.api.Test;
+
+public class SingerConfigSchemaTest {
+
+  @Test
+  void testFile() throws IOException {
+    final String schema = Files.readString(SingerConfigSchema.SINGER_MESSAGE.getFile().toPath(), StandardCharsets.UTF_8);
+    assertTrue(schema.contains("title"));
+  }
+
+  @Test
+  void testPrepareKnownSchemas() {
+    for (SingerConfigSchema value : SingerConfigSchema.values()) {
+      assertTrue(Files.exists(value.getFile().toPath()));
+    }
+  }
+
+}

dataline-singer/src/main/resources/json/SingerMessage.json renamed to dataline-singer/src/main/resources/singer_json/SingerMessage.json

@@ -48,8 +48,18 @@ public class SingerJsonStreamFactory implements StreamFactory {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(SingerJsonStreamFactory.class);
 
+  private final SingerProtocolPredicate singerProtocolValidator;
+
+  public SingerJsonStreamFactory() {
+    this(new SingerProtocolPredicate());
+  }
+
+  SingerJsonStreamFactory(SingerProtocolPredicate singerProtocolPredicate) {
+    singerProtocolValidator = singerProtocolPredicate;
+  }
+
   public Stream<SingerMessage> create(BufferedReader bufferedReader) {
-    return bufferedReader.lines().map(this::parseJsonOrNull).filter(Objects::nonNull);
+    return bufferedReader.lines().filter(singerProtocolValidator).map(this::parseJsonOrNull).filter(Objects::nonNull);
   }
 
   private SingerMessage parseJsonOrNull(String record) {

dataline-singer/src/test/java/io/dataline/singer/SingerConfigSchemaTest.java

@@ -0,0 +1,48 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Dataline
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.dataline.workers.protocol.singer;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.dataline.commons.json.JsonSchemaValidator;
+import io.dataline.commons.json.Jsons;
+import io.dataline.singer.SingerConfigSchema;
+import java.util.function.Predicate;
+
+public class SingerProtocolPredicate implements Predicate<String> {
+
+  private final JsonSchemaValidator jsonSchemaValidator;
+  private final JsonNode schema;
+
+  public SingerProtocolPredicate() {
+    jsonSchemaValidator = new JsonSchemaValidator();
+    schema = JsonSchemaValidator.getSchema(SingerConfigSchema.SINGER_MESSAGE.getFile());
+  }
+
+  @Override
+  public boolean test(String s) {
+    return jsonSchemaValidator.test(schema, Jsons.deserialize(s));
+  }
+
+}

dataline-workers/src/main/java/io/dataline/workers/protocol/singer/SingerJsonStreamFactory.java

@@ -25,6 +25,9 @@
 package io.dataline.workers.protocol.singer;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import io.dataline.commons.json.Jsons;
 import io.dataline.singer.SingerMessage;
@@ -34,6 +37,7 @@
 import java.io.InputStreamReader;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 @SuppressWarnings("StringBufferReplaceableByString")
@@ -42,6 +46,14 @@ class SingerJsonStreamFactoryTest {
   private static final String TABLE_NAME = "user_preferences";
   private static final String COLUMN_NAME = "favorite_color";
 
+  private SingerProtocolPredicate singerProtocolPredicate;
+
+  @BeforeEach
+  public void setup() {
+    singerProtocolPredicate = mock(SingerProtocolPredicate.class);
+    when(singerProtocolPredicate.test(any())).thenReturn(true);
+  }
+
   @Test
   public void testValid() {
     final SingerMessage record1 =
@@ -63,55 +75,37 @@ public void testValid() {
         expectedStream.collect(Collectors.toList()), messageStream.collect(Collectors.toList()));
   }
 
-  // TODO these test cases are commented out because jsonschema2pojo does not generate annotations
-  // that allow us to validate records correctly.
-  // Until we add validation, deserializing invalid records (e.g: ones missing required fields) will
-  // succeed, so we temporarily comment these cases out.
-  // @Test
-  // public void testInvalid() {
-  // final SingerMessage record1 =
-  // MessageUtils.createRecordMessage(TABLE_NAME, COLUMN_NAME, "green");
-  // final SingerMessage record2 =
-  // MessageUtils.createRecordMessage(TABLE_NAME, COLUMN_NAME, "yellow");
-  //
-  // final String inputString =
-  // new StringBuilder()
-  // .append(Jsons.serialize(record1))
-  // .append('\n')
-  // .append("{ \"fish\": \"tuna\"}")
-  // .append('\n')
-  // .append(Jsons.serialize(record2))
-  // .toString();
-  //
-  // final Stream<SingerMessage> messageStream = stringToSingerMessageStream(inputString);
-  // final Stream<SingerMessage> expectedStream = Stream.of(record1, record2);
-  //
-  // assertEquals(
-  // expectedStream.collect(Collectors.toList()), messageStream.collect(Collectors.toList()));
-  // }
-
   @Test
-  public void testMissingNewLineBetweenValidRecords() {
+  public void testInvalid() {
+
     final SingerMessage record1 =
         MessageUtils.createRecordMessage(TABLE_NAME, COLUMN_NAME, "green");
     final SingerMessage record2 =
         MessageUtils.createRecordMessage(TABLE_NAME, COLUMN_NAME, "yellow");
+    final String invalidRecord = "{ \"fish\": \"tuna\"}";
+
+    when(singerProtocolPredicate.test(Jsons.serialize(record1))).thenReturn(true);
+    when(singerProtocolPredicate.test(Jsons.serialize(record2))).thenReturn(true);
+    when(singerProtocolPredicate.test(invalidRecord)).thenReturn(false);
 
     final String inputString =
         new StringBuilder()
             .append(Jsons.serialize(record1))
+            .append('\n')
+            .append("{ \"fish\": \"tuna\"}")
+            .append('\n')
             .append(Jsons.serialize(record2))
             .toString();
 
     final Stream<SingerMessage> messageStream = stringToSingerMessageStream(inputString);
-    final Stream<SingerMessage> expectedStream = Stream.of(record1);
+    final Stream<SingerMessage> expectedStream = Stream.of(record1, record2);
 
     assertEquals(
         expectedStream.collect(Collectors.toList()), messageStream.collect(Collectors.toList()));
   }
 
   @Test
-  public void testMissingNewLineAndLineStartsWithValidRecord() {
+  public void testMissingNewLineBetweenValidRecords() {
     final SingerMessage record1 =
         MessageUtils.createRecordMessage(TABLE_NAME, COLUMN_NAME, "green");
     final SingerMessage record2 =
@@ -120,44 +114,20 @@ public void testMissingNewLineAndLineStartsWithValidRecord() {
     final String inputString =
         new StringBuilder()
             .append(Jsons.serialize(record1))
-            .append("{ \"fish\": \"tuna\"}")
-            .append('\n')
             .append(Jsons.serialize(record2))
             .toString();
 
     final Stream<SingerMessage> messageStream = stringToSingerMessageStream(inputString);
-    final Stream<SingerMessage> expectedStream = Stream.of(record1, record2);
+    final Stream<SingerMessage> expectedStream = Stream.of(record1);
 
     assertEquals(
         expectedStream.collect(Collectors.toList()), messageStream.collect(Collectors.toList()));
   }
 
-  // @Test
-  // public void testMissingNewLineAndLineStartsWithInvalidRecord() {
-  // final SingerMessage record1 =
-  // MessageUtils.createRecordMessage(TABLE_NAME, COLUMN_NAME, "green");
-  // final SingerMessage record2 =
-  // MessageUtils.createRecordMessage(TABLE_NAME, COLUMN_NAME, "yellow");
-  //
-  // final String inputString =
-  // new StringBuilder()
-  // .append(Jsons.serialize(record1))
-  // .append('\n')
-  // .append("{ \"fish\": \"tuna\"}")
-  // .append(Jsons.serialize(record2))
-  // .toString();
-  //
-  // final Stream<SingerMessage> messageStream = stringToSingerMessageStream(inputString);
-  // final Stream<SingerMessage> expectedStream = Stream.of(record1);
-  //
-  // assertEquals(
-  // expectedStream.collect(Collectors.toList()), messageStream.collect(Collectors.toList()));
-  // }
-
-  private static Stream<SingerMessage> stringToSingerMessageStream(String inputString) {
+  private Stream<SingerMessage> stringToSingerMessageStream(String inputString) {
     InputStream inputStream = new ByteArrayInputStream(inputString.getBytes());
     final BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
-    return new SingerJsonStreamFactory().create(bufferedReader);
+    return new SingerJsonStreamFactory(singerProtocolPredicate).create(bufferedReader);
   }
 
 }