Handle exceptions in KafkaSplitManager
@@ -20,7 +20,6 @@ import com.facebook.presto.spi.ConnectorTableLayoutHandle;
import com.facebook.presto.spi.FixedSplitSource;
import com.facebook.presto.spi.HostAddress;
import com.facebook.presto.spi.PrestoException;
-import com.facebook.presto.spi.StandardErrorCode;
import com.facebook.presto.spi.connector.ConnectorSplitManager;
import com.facebook.presto.spi.connector.ConnectorTransactionHandle;
import com.google.common.collect.ImmutableList;
@@ -46,6 +45,7 @@ import java.util.concurrent.ThreadLocalRandom;
import static com.facebook.presto.kafka.KafkaErrorCode.KAFKA_SPLIT_ERROR;
import static com.facebook.presto.kafka.KafkaHandleResolver.convertLayout;
+import static com.facebook.presto.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
@@ -78,7 +78,7 @@ public class KafkaSplitManager
public ConnectorSplitSource getSplits(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorTableLayoutHandle layout, SplitSchedulingStrategy splitSchedulingStrategy)
{
KafkaTableHandle kafkaTableHandle = convertLayout(layout).getTable();
-
+ try {
SimpleConsumer simpleConsumer = consumerManager.getConsumer(selectRandom(nodes));
TopicMetadataRequest topicMetadataRequest = new TopicMetadataRequest(ImmutableList.of(kafkaTableHandle.getTopicName()));
@@ -92,7 +92,7 @@ public class KafkaSplitManager
Broker leader = part.leader();
if (leader == null) {
- throw new PrestoException(StandardErrorCode.GENERIC_INTERNAL_ERROR, format("Leader election in progress for Kafka topic '%s' partition %s", metadata.topic(), part.partitionId()));
+ throw new PrestoException(GENERIC_INTERNAL_ERROR, format("Leader election in progress for Kafka topic '%s' partition %s", metadata.topic(), part.partitionId()));
}
HostAddress partitionLeader = HostAddress.fromParts(leader.host(), leader.port());
@@ -119,6 +119,13 @@ public class KafkaSplitManager
return new FixedSplitSource(splits.build());
}
+ 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 list splits for table '%s' reading topic '%s'", kafkaTableHandle.getTableName(), kafkaTableHandle.getTopicName()), e);
+ }
+ }
private static long[] findAllOffsets(SimpleConsumer consumer, String topicName, int partitionId)
{
@@ -136,7 +143,7 @@ public class KafkaSplitManager
if (offsetResponse.hasError()) {
short errorCode = offsetResponse.errorCode(topicName, partitionId);
log.warn("Offset 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 + "'");
}
return offsetResponse.offsets(topicName, partitionId);
