Merge pull request #332 from liquidm/improve-kafka

improve kafka intake stability

kafka-eight/src/main/java/io/druid/firehose/kafka/KafkaEightFirehoseFactory.java

@@ -107,7 +107,15 @@ public class KafkaEightFirehoseFactory implements FirehoseFactory
           return null;
         }
 
-        return parser.parse(ByteBuffer.wrap(message));
+        try {
+          return parser.parse(ByteBuffer.wrap(message));
+        }
+        catch (Exception e) {
+          throw new FormattedException.Builder()
+              .withErrorCode(FormattedException.ErrorCode.UNPARSABLE_ROW)
+              .withMessage(String.format("Error parsing[%s], got [%s]", ByteBuffer.wrap(message), e.toString()))
+              .build();
+        }
       }
 
       @Override

kafka-seven/src/main/java/io/druid/firehose/kafka/KafkaSevenFirehoseFactory.java

@@ -120,7 +120,15 @@ public class KafkaSevenFirehoseFactory implements FirehoseFactory
 
       public InputRow parseMessage(Message message) throws FormattedException
       {
-        return parser.parse(message.payload());
+        try {
+          return parser.parse(message.payload());
+        }
+        catch (Exception e) {
+          throw new FormattedException.Builder()
+              .withErrorCode(FormattedException.ErrorCode.UNPARSABLE_ROW)
+              .withMessage(String.format("Error parsing[%s], got [%s]", message.payload(), e.toString()))
+              .build();
+        }
       }
 
       @Override