SUMOK-23 Transformation errors handling properly

Author: Juan Pablo Diaz-Vaz

src/main/java/com/mcplusa/sumologic/CloudWatchMessageModelSumologicTransformer.java

@@ -12,6 +12,9 @@
 import java.io.ByteArrayInputStream;
 import java.io.BufferedReader;
 import java.io.InputStreamReader;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetDecoder;
 import java.util.zip.GZIPInputStream;
 
 /**
@@ -39,6 +42,12 @@ public String fromClass(SimpleKinesisMessageModel message) {
     public SimpleKinesisMessageModel toClass(Record record) throws IOException {
       byte[] decodedRecord = record.getData().array();
       String stringifiedRecord = decompressGzip(decodedRecord);
+      
+      if (stringifiedRecord == null) {
+        LOG.error("Unable to decompress the record: "+new String(record.getData().array()));
+        LOG.error("Not attempting to transform into a Message Model");
+        return null;
+      }
 
       return new SimpleKinesisMessageModel(stringifiedRecord);
     }
@@ -59,5 +68,19 @@ public static String decompressGzip(byte[] compressedData) {
         return null;
       }
     }
+    
+    public static String byteBufferToString(ByteBuffer buffer){
+      String data = "";
+      CharsetDecoder decoder = Charset.forName("UTF-8").newDecoder();
+      try{
+        int old_position = buffer.position();
+        data = decoder.decode(buffer).toString();
+        buffer.position(old_position);  
+      }catch (Exception e){
+        e.printStackTrace();
+        return "";
+      }
+      return data;
+    }
 
 }

src/test/java/com/mcplusa/sumologic/CloudWatchMessageModelSumologicTransformerTest.java

@@ -0,0 +1,48 @@
+package com.mcplusa.sumologic;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.Ignore;
+
+import com.amazonaws.services.kinesis.model.Record;
+import com.mcplusa.sumologic.CloudWatchMessageModelSumologicTransformer;
+import com.mcplusa.sumologic.SimpleKinesisMessageModel;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetEncoder;
+import java.nio.CharBuffer;
+import java.nio.ByteBuffer;
+
+
+public class CloudWatchMessageModelSumologicTransformerTest {
+  public static Charset charset = Charset.forName("UTF-8");
+  public static CharsetEncoder encoder = charset.newEncoder();
+  
+  @Test
+  public void theTransformerShouldFailGracefullyWhenUnableToTransform () {
+    CloudWatchMessageModelSumologicTransformer transfomer = new CloudWatchMessageModelSumologicTransformer();
+    
+    String randomData = "Some random string without GZIP compression";
+    ByteBuffer bufferedData = null;
+    try {
+      bufferedData = encoder.encode(CharBuffer.wrap(randomData));
+    } catch (Exception e) {
+      Assert.fail("Getting error: "+e.getMessage());
+    }
+    
+    Record mockedRecord = new Record();
+    mockedRecord.setData(bufferedData);
+    
+    SimpleKinesisMessageModel messageModel = null;
+    try {
+      messageModel = transfomer.toClass(mockedRecord);
+    } catch (IOException e) {
+      Assert.fail("Getting error while transforming: "+e.getMessage());
+    }
+    
+    Assert.assertNull(messageModel);
+  }
+}