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);
    }
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.
@return