Better exception handling in KafkaRecordSet
@@ -43,6 +43,7 @@ import static com.facebook.presto.decoder.FieldValueProviders.bytesValueProvider
import static com.facebook.presto.decoder.FieldValueProviders.longValueProvider;
import static com.facebook.presto.kafka.KafkaErrorCode.KAFKA_SPLIT_ERROR;
import static com.google.common.base.Preconditions.checkArgument;
+import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
/**
@@ -307,6 +308,7 @@ public class KafkaRecordSet
private void openFetchRequest()
{
+ try {
if (messageAndOffsetIterator == null) {
log.debug("Fetching %d bytes from offset %d (%d - %d). %d messages read so far", KAFKA_READ_BUFFER_SIZE, cursorOffset, split.getStart(), split.getEnd(), totalMessages);
FetchRequest req = new FetchRequestBuilder()
@@ -322,11 +324,27 @@ public class KafkaRecordSet
if (fetchResponse.hasError()) {
short errorCode = fetchResponse.errorCode(split.getTopicName(), split.getPartitionId());
log.warn("Fetch response has error: %d", errorCode);
- throw new PrestoException(KAFKA_SPLIT_ERROR, "could not fetch data from Kafka, error code is '" + errorCode + "'");
+ throw new RuntimeException("could not fetch data from Kafka, error code is '" + errorCode + "'");
}
messageAndOffsetIterator = fetchResponse.messageSet(split.getTopicName(), split.getPartitionId()).iterator();
}
}
+ catch (Exception e) { // Catch all exceptions because Kafka library is written in scala and checked exceptions are not declared in method signature.
+ if (e instanceof PrestoException) {
+ throw e;
+ }
+ throw new PrestoException(
+ KAFKA_SPLIT_ERROR,
+ format(
+ "Cannot read data from topic '%s', partition '%s', startOffset %s, endOffset %s, leader %s ",
+ split.getTopicName(),
+ split.getPartitionId(),
+ split.getStart(),
+ split.getEnd(),
+ split.getLeader()),
+ e);
+ }
+ }
}
}
