Apache
/
druid
mirror of https://github.com/apache/druid.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

improve kafka intake stability

This commit is contained in:
Hagen Rother 2013-12-17 15:27:17 +01:00
parent 9fc0e18962
commit af202d7576
2 changed files with 18 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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

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

@ -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