presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaSplitManager.java/KafkaSplitManager/getSplits
class KafkaSplitManager:    @Override
    public ConnectorSplitSource getSplits(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorTableHandle table, SplitSchedulingStrategy splitSchedulingStrategy)
    {
        KafkaTableHandle kafkaTableHandle = (KafkaTableHandle) table;
        try {
            SimpleConsumer simpleConsumer = consumerManager.getConsumer(selectRandom(nodes));

            TopicMetadataRequest topicMetadataRequest = new TopicMetadataRequest(ImmutableList.of(kafkaTableHandle.getTopicName()));
            TopicMetadataResponse topicMetadataResponse = simpleConsumer.send(topicMetadataRequest);

            ImmutableList.Builder<ConnectorSplit> splits = ImmutableList.builder();

            for (TopicMetadata metadata : topicMetadataResponse.topicsMetadata()) {
                for (PartitionMetadata part : metadata.partitionsMetadata()) {
                    log.debug("Adding Partition %s/%s", metadata.topic(), part.partitionId());

                    BrokerEndPoint leader = part.leader();
                    if (leader == null) {
                        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());

                    SimpleConsumer leaderConsumer = consumerManager.getConsumer(partitionLeader);
                    // Kafka contains a reverse list of "end - start" pairs for the splits

                    long[] offsets = findAllOffsets(leaderConsumer, metadata.topic(), part.partitionId());

                    for (int i = offsets.length - 1; i > 0; i--) {
                        KafkaSplit split = new KafkaSplit(
                                metadata.topic(),
                                kafkaTableHandle.getKeyDataFormat(),
                                kafkaTableHandle.getMessageDataFormat(),
                                kafkaTableHandle.getKeyDataSchemaLocation().map(KafkaSplitManager::readSchema),
                                kafkaTableHandle.getMessageDataSchemaLocation().map(KafkaSplitManager::readSchema),
                                part.partitionId(),
                                offsets[i],
                                offsets[i - 1],
                                partitionLeader);
                        splits.add(split);
                    }
                }
            }

            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);
        }
    }