private org . apache . kafka . clients . consumer . KafkaConsumer < Object , Object > createConsumer ( ) { Map < String , Object > consumerProps = new HashMap <> ( ) ; consumerProps . put ( ConsumerConfig . CLIENT_ID_CONFIG , Optional . ofNullable ( endpointConfiguration . getClientId ( ) ) . orElse ( KafkaMessageHeaders . KAFKA_PREFIX + "consumer_" + UUID . randomUUID ( ) . toString ( ) ) ) ; consumerProps . put ( ConsumerConfig . GROUP_ID_CONFIG , endpointConfiguration . getConsumerGroup ( ) ) ; consumerProps . put ( ConsumerConfig . BOOTSTRAP_SERVERS_CONFIG , Optional . ofNullable ( endpointConfiguration . getServer ( ) ) . orElse ( "localhost:9092" ) ) ; consumerProps . put ( ConsumerConfig . MAX_POLL_RECORDS_CONFIG , 1 ) ; consumerProps . put ( ConsumerConfig . ENABLE_AUTO_COMMIT_CONFIG , endpointConfiguration . isAutoCommit ( ) ) ; consumerProps . put ( ConsumerConfig . AUTO_COMMIT_INTERVAL_MS_CONFIG , endpointConfiguration . getAutoCommitInterval ( ) ) ; consumerProps . put ( ConsumerConfig . AUTO_OFFSET_RESET_CONFIG , endpointConfiguration . getOffsetReset ( ) ) ; consumerProps . put ( ConsumerConfig . KEY_DESERIALIZER_CLASS_CONFIG , endpointConfiguration . getKeyDeserializer ( ) ) ; consumerProps . put ( ConsumerConfig . VALUE_DESERIALIZER_CLASS_CONFIG , endpointConfiguration . getValueDeserializer ( ) ) ; consumerProps . putAll ( endpointConfiguration . getConsumerProperties ( ) ) ; return new org . apache . kafka . clients . consumer . KafkaConsumer <> ( consumerProps ) ; }
Create new Kafka consumer with given endpoint configuration .