Package com.amazonaws.athena.connectors.kafka

Class KafkaUtils

  • public class KafkaUtils
extends Object

    • Method Detail

      • getKafkaConsumer

        public static org.apache.kafka.clients.consumer.Consumer<String,​String> getKafkaConsumer​(Map<String,​String> configOptions)
                                                                                        throws Exception
        Creates Kafka consumer instance.
        Returns:
        Throws:
        Exception

      • getKafkaConsumer

        public static org.apache.kafka.clients.consumer.Consumer<String,​TopicResultSet> getKafkaConsumer​(org.apache.arrow.vector.types.pojo.Schema schema,
                                                                                                       Map<String,​String> configOptions)
                                                                                                throws Exception
        Creates instance of Kafka consumer. In the properties we have to specify the Deserializer type, we are supporting only JSON or CSV data from kafka topic. Therefor, to transform topic data into pojo we need to specify Deserializer type to KafkaConsumer. Schema metadata can tell use about the topic data type i.e. dataFormat = json | csv.
        Parameters:
        schema - - instance of Schema
        Returns:
        Consumer Consumer
        Throws:
        Exception - - Exception

      • getAvroKafkaConsumer

        public static org.apache.kafka.clients.consumer.Consumer<String,​org.apache.avro.generic.GenericRecord> getAvroKafkaConsumer​(Map<String,​String> configOptions)
                                                                                                                           throws Exception
        Throws:
        Exception

      • getProtobufKafkaConsumer

        public static org.apache.kafka.clients.consumer.Consumer<String,​com.google.protobuf.DynamicMessage> getProtobufKafkaConsumer​(Map<String,​String> configOptions)
                                                                                                                            throws Exception
        Throws:
        Exception

      • setSaslPlainAuthKafkaProperties

        protected static Properties setSaslPlainAuthKafkaProperties​(Properties properties,
                                                            Map<String,​String> configOptions)
                                                     throws Exception
        Creates the required SASL based settings for kafka consumer.
        Parameters:
        properties - - SASL/PLAINTEXT properties for kafka consumer
        Returns:
        Properties
        Throws:
        Exception - - Exception

      • copyCertificatesFromS3ToTempFolder

        protected static Path copyCertificatesFromS3ToTempFolder​(Map<String,​String> configOptions)
                                                  throws Exception
        Downloads the truststore and keystore certificates from S3 to temp directory.
        Throws:
        Exception - - Exception

      • createSplitParam

        public static SplitParameters createSplitParam​(Map<String,​String> params)
        Translates Split parameters as readable pojo format.
        Parameters:
        params - - the properties for split object
        Returns:
        SplitParameters

      • toArrowType

        public static org.apache.arrow.vector.types.pojo.ArrowType toArrowType​(String dataType)
        Converts string data type name to ArrowType. After pulling schema from glue schema registry we use this method to convert the data type in glue schema to ArrowType.
        Parameters:
        dataType - - the column name of schema
        Returns:
        ArrowType