@Override
    protected void initQueue(T queue, QueueSpec spec) throws Exception {
        queue.setKafkaClient(defaultKafkaClient).setProducerType(defaultProducerType)
                .setKafkaProducerProperties(defaultProducerProps)
                .setKafkaConsumerProperties(defaultConsumerProps);
        queue.setSendAsync(defaultSendAsync);

        String bootstrapServers = spec.getField(SPEC_FIELD_BOOTSTRAP_SERVERS);
        bootstrapServers = StringUtils.isBlank(bootstrapServers) ? defaultBootstrapServers
                : bootstrapServers;
        if (StringUtils.isBlank(bootstrapServers)) {
            throw new IllegalArgumentException(
                    "Empty or Invalid value for param [" + SPEC_FIELD_BOOTSTRAP_SERVERS + "]!");
        }
        queue.setKafkaBootstrapServers(bootstrapServers);

        String consumerGroupId = spec.getField(SPEC_FIELD_CONSUMER_GROUP_ID);
        consumerGroupId = StringUtils.isBlank(consumerGroupId) ? defaultConsumerGroupId
                : consumerGroupId;
        if (!StringUtils.isBlank(consumerGroupId)) {
            queue.setConsumerGroupId(consumerGroupId);
        }

        String topicName = spec.getField(SPEC_FIELD_TOPIC);
        topicName = StringUtils.isBlank(topicName) ? defaultTopicName : topicName;
        if (StringUtils.isBlank(topicName)) {
            throw new IllegalArgumentException(
                    "Empty or Invalid value for param [" + SPEC_FIELD_TOPIC + "]!");
        }
        queue.setTopicName(topicName);

        String producerTypeStr = spec.getField(SPEC_FIELD_PRODUCER_TYPE);
        if (!StringUtils.isBlank(producerTypeStr)) {
            try {
                ProducerType producerType = ProducerType.valueOf(producerTypeStr);
                if (producerType != null) {
                    queue.setProducerType(producerType);
                }
            } catch (Exception e) {
            }
        }

        Properties producerProps = spec.getField(SPEC_FIELD_PRODUCER_PROPERTIES, Properties.class);
        if (producerProps != null) {
            queue.setKafkaProducerProperties(producerProps);
        }

        Properties consumerProps = spec.getField(SPEC_FIELD_CONSUMER_PROPERTIES, Properties.class);
        if (consumerProps != null) {
            queue.setKafkaConsumerProperties(consumerProps);
        }

        Boolean sendAsync = spec.getField(SPEC_FIELD_SEND_ASYNC, Boolean.class);
        if (sendAsync != null) {
            queue.setSendAsync(sendAsync.booleanValue());
        }

        super.initQueue(queue, spec);
    }
@ Override protected void initQueue ( T queue , QueueSpec spec ) throws Exception { queue . setKafkaClient ( defaultKafkaClient ) . setProducerType ( defaultProducerType ) . setKafkaProducerProperties ( defaultProducerProps ) . setKafkaConsumerProperties ( defaultConsumerProps ) ; queue . setSendAsync ( defaultSendAsync ) ; String bootstrapServers = spec . getField ( SPEC_FIELD_BOOTSTRAP_SERVERS ) ; bootstrapServers = StringUtils . isBlank ( bootstrapServers ) ? defaultBootstrapServers : bootstrapServers ; if ( StringUtils . isBlank ( bootstrapServers ) ) { throw new IllegalArgumentException ( "Empty or Invalid value for param [" + SPEC_FIELD_BOOTSTRAP_SERVERS + "]!" ) ; } queue . setKafkaBootstrapServers ( bootstrapServers ) ; String consumerGroupId = spec . getField ( SPEC_FIELD_CONSUMER_GROUP_ID ) ; consumerGroupId = StringUtils . isBlank ( consumerGroupId ) ? defaultConsumerGroupId : consumerGroupId ; if ( ! StringUtils . isBlank ( consumerGroupId ) ) { queue . setConsumerGroupId ( consumerGroupId ) ; } String topicName = spec . getField ( SPEC_FIELD_TOPIC ) ; topicName = StringUtils . isBlank ( topicName ) ? defaultTopicName : topicName ; if ( StringUtils . isBlank ( topicName ) ) { throw new IllegalArgumentException ( "Empty or Invalid value for param [" + SPEC_FIELD_TOPIC + "]!" ) ; } queue . setTopicName ( topicName ) ; String producerTypeStr = spec . getField ( SPEC_FIELD_PRODUCER_TYPE ) ; if ( ! StringUtils . isBlank ( producerTypeStr ) ) { try { ProducerType producerType = ProducerType . valueOf ( producerTypeStr ) ; if ( producerType != null ) { queue . setProducerType ( producerType ) ; } } catch ( Exception e ) { } } Properties producerProps = spec . getField ( SPEC_FIELD_PRODUCER_PROPERTIES , Properties . class ) ; if ( producerProps != null ) { queue . setKafkaProducerProperties ( producerProps ) ; } Properties consumerProps = spec . getField ( SPEC_FIELD_CONSUMER_PROPERTIES , Properties . class ) ; if ( consumerProps != null ) { queue . setKafkaConsumerProperties ( consumerProps ) ; } Boolean sendAsync = spec . getField ( SPEC_FIELD_SEND_ASYNC , Boolean . class ) ; if ( sendAsync != null ) { queue . setSendAsync ( sendAsync . booleanValue ( ) ) ; } super . initQueue ( queue , spec ) ; }
{@inheritDoc}

@throws Exception