test: add tests for set errorType for UnsupportedOperationException to INVALID_MESSAGE_ERROR (#47)

Author: sumitaich1998

src/test/java/io/odpf/depot/redis/parsers/RedisParserTest.java

@@ -69,11 +69,13 @@ public void setup() throws IOException {
         TestMessage message3 = TestMessage.newBuilder().setOrderNumber("test-order-3").setOrderDetails("ORDER-DETAILS-3").build();
         TestMessage message4 = TestMessage.newBuilder().setOrderNumber("test-order-4").setOrderDetails("ORDER-DETAILS-4").build();
         TestMessage message5 = TestMessage.newBuilder().setOrderNumber("test-order-5").setOrderDetails("ORDER-DETAILS-5").build();
+        TestMessage message6 = TestMessage.newBuilder().setOrderNumber("test-order-6").setOrderDetails("ORDER-DETAILS-6").build();
         messages.add(new OdpfMessage(null, message1.toByteArray()));
         messages.add(new OdpfMessage(null, message2.toByteArray()));
         messages.add(new OdpfMessage(null, message3.toByteArray()));
         messages.add(new OdpfMessage(null, message4.toByteArray()));
         messages.add(new OdpfMessage(null, message5.toByteArray()));
+        messages.add(new OdpfMessage(null, message6.toByteArray()));
     }
 
     public void setupParserResponse() throws IOException {
@@ -96,14 +98,15 @@ public void shouldConvertOdpfMessageToRedisRecords() throws IOException {
         Map<Boolean, List<RedisRecord>> splitterRecords = parsedRecords.stream().collect(Collectors.partitioningBy(RedisRecord::isValid));
         List<RedisRecord> invalidRecords = splitterRecords.get(Boolean.FALSE);
         List<RedisRecord> validRecords = splitterRecords.get(Boolean.TRUE);
-        assertEquals(5, validRecords.size());
+        assertEquals(6, validRecords.size());
         assertTrue(invalidRecords.isEmpty());
         List<RedisRecord> expectedRecords = new ArrayList<>();
         expectedRecords.add(new RedisRecord(new RedisKeyValueEntry("test-key", "test-order-1", null), 0L, null, "{}", true));
         expectedRecords.add(new RedisRecord(new RedisKeyValueEntry("test-key", "test-order-2", null), 1L, null, "{}", true));
         expectedRecords.add(new RedisRecord(new RedisKeyValueEntry("test-key", "test-order-3", null), 2L, null, "{}", true));
         expectedRecords.add(new RedisRecord(new RedisKeyValueEntry("test-key", "test-order-4", null), 3L, null, "{}", true));
         expectedRecords.add(new RedisRecord(new RedisKeyValueEntry("test-key", "test-order-5", null), 4L, null, "{}", true));
+        expectedRecords.add(new RedisRecord(new RedisKeyValueEntry("test-key", "test-order-6", null), 5L, null, "{}", true));
         IntStream.range(0, expectedRecords.size()).forEach(index -> assertEquals(expectedRecords.get(index).toString(), parsedRecords.get(index).toString()));
     }
 
@@ -113,14 +116,16 @@ public void shouldReportValidAndInvalidRecords() throws IOException {
         when(odpfMessageParser.parse(messages.get(2), SinkConnectorSchemaMessageMode.LOG_MESSAGE, schemaClass)).thenThrow(new IOException("Error while parsing protobuf"));
         when(odpfMessageParser.parse(messages.get(3), SinkConnectorSchemaMessageMode.LOG_MESSAGE, schemaClass)).thenThrow(new ConfigurationException("Invalid field config : INVALID"));
         when(odpfMessageParser.parse(messages.get(4), SinkConnectorSchemaMessageMode.LOG_MESSAGE, schemaClass)).thenThrow(new IllegalArgumentException("Config REDIS_CONFIG is empty"));
+        when(odpfMessageParser.parse(messages.get(5), SinkConnectorSchemaMessageMode.LOG_MESSAGE, schemaClass)).thenThrow(new UnsupportedOperationException("some message"));
         List<RedisRecord> parsedRecords = redisParser.convert(messages);
         Map<Boolean, List<RedisRecord>> splitterRecords = parsedRecords.stream().collect(Collectors.partitioningBy(RedisRecord::isValid));
         List<RedisRecord> invalidRecords = splitterRecords.get(Boolean.FALSE);
         List<RedisRecord> validRecords = splitterRecords.get(Boolean.TRUE);
         assertEquals(2, validRecords.size());
-        assertEquals(3, invalidRecords.size());
+        assertEquals(4, invalidRecords.size());
         assertEquals(ErrorType.DESERIALIZATION_ERROR, parsedRecords.get(2).getErrorInfo().getErrorType());
         assertEquals(ErrorType.UNKNOWN_FIELDS_ERROR, parsedRecords.get(3).getErrorInfo().getErrorType());
         assertEquals(ErrorType.DEFAULT_ERROR, parsedRecords.get(4).getErrorInfo().getErrorType());
+        assertEquals(ErrorType.INVALID_MESSAGE_ERROR, parsedRecords.get(5).getErrorInfo().getErrorType());
     }
 }