0

我正在尝试使用来自 logstash 的 IBM Message Hub。我的logstash.conf：

input {
    kafka {
        bootstrap_servers =>  "kafka04-prod01.messagehub.services.eu-de.bluemix.net:9093,kafka05-prod01.messagehub.services.eu-de.bluemix.net:9093,kafka01-prod01.messagehub.services.eu-de.bluemix.net:9093,kafka03-prod01.messagehub.services.eu-de.bluemix.net:9093,kafka02-prod01.messagehub.services.eu-de.bluemix.net:9093"
        security_protocol => "SASL_SSL"
        sasl_mechanism => "PLAIN"
        jaas_path => "kafka_jaas.conf"
        ssl => true
        topics => [
                "transactions_load"
            ]
    }
}
output {
    stdout { }
}

基于Message Hub Consumer 示例，我缺少以下 kafka 配置属性：

ssl.protocol=TLSv1.2
ssl.enabled.protocols=TLSv1.2
ssl.endpoint.identification.algorithm=HTTPS

但是，我无法从logstash 文档中看到如何设置这些。

我的 kafka_jaas.conf 文件如下所示：

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  serviceName="kafka"
  username="****"
  password="****";
  };

结果是

[2017-10-08T10:00:48,325][ERROR][logstash.inputs.kafka    ] Unable to create Kafka consumer from given configuration {:kafka_error_message=>java.lang.NullPointerException, :cause=>nil}
....
[2017-10-08T10:00:52,717][ERROR][logstash.pipeline        ] A plugin had an unrecoverable error. Will restart this plugin.
  Plugin: <LogStash::Inputs::Kafka bootstrap_servers=>"kafka04-prod01.messagehub.services.eu-de.bluemix.net:9093,kafka05-prod01.messagehub.services.eu-de.bluemix.net:9093,kafka01-prod01.messagehub.services.eu-de.bluemix.net:9093,kafka03-prod01.messagehub.services.eu-de.bluemix.net:9093,kafka02-prod01.messagehub.services.eu-de.bluemix.net:9093", 
security_protocol=>"SASL_SSL", 
sasl_mechanism=>"PLAIN", 
jaas_path=>"kafka_jaas.conf", 
ssl=>true, 
topics=>["transactions_load"], 
id=>"xxxxx",  
enable_metric=>true, 
codec=><LogStash::Codecs::Plain id=>"xxxxx", 
enable_metric=>true, 
charset=>"UTF-8">, 
auto_commit_interval_ms=>"5000", 
client_id=>"logstash", 
consumer_threads=>1, 
enable_auto_commit=>"true", 
group_id=>"logstash",
key_deserializer_class=>
   "org.apache.kafka.common.serialization.StringDeserializer", 
value_deserializer_class=>
   "org.apache.kafka.common.serialization.StringDeserializer", 
poll_timeout_ms=>100, 
decorate_events=>false>
  Error: uncaught throw  in thread 0x301b0

如果您有 Message Hub 帐户，则可以简单地复制我的问题：

wget https://artifacts.elastic.co/downloads/logstash/logstash-5.4.2.tar.gz
tar xvzf logstash-5.4.2.tar.gz
# create logstash.conf from above
# create kafka_jaas.conf from above
./logstash-5.4.2/bin/logstash -f logstash.conf

（最新的logstash也有同样的问题，目前是5.6.2）

4

1 回答 1

1

我的解决方案是：

# replace old kafka consumer jar files

rm logstash-5.4.2/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-5.1.7/vendor/jar-dependencies/runtime-jars/kafka-clients-0.10.0.1.jar

cp ~/.m2/repository/org/apache/kafka/kafka-clients/0.10.2.1/kafka-clients-0.10.2.1.jar logstash-5.4.2/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-5.1.7/vendor/jar-dependencies/runtime-jars/

然后编辑kafka.rb文件：

# logstash-5.4.2/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-5.1.7/lib/logstash/inputs/kafka.rb

  private
  def create_consumer(client_id)
    begin
      props = java.util.Properties.new
      kafka = org.apache.kafka.clients.consumer.ConsumerConfig

      props.put(kafka::AUTO_COMMIT_INTERVAL_MS_CONFIG, auto_commit_interval_ms)
      props.put(kafka::AUTO_OFFSET_RESET_CONFIG, auto_offset_reset) unless auto_offset_reset.nil?
      props.put(kafka::BOOTSTRAP_SERVERS_CONFIG, bootstrap_servers)
      props.put(kafka::CHECK_CRCS_CONFIG, check_crcs) unless check_crcs.nil?
      props.put(kafka::CLIENT_ID_CONFIG, client_id)
      props.put(kafka::CONNECTIONS_MAX_IDLE_MS_CONFIG, connections_max_idle_ms) unless connections_max_idle_ms.nil?
      props.put(kafka::ENABLE_AUTO_COMMIT_CONFIG, enable_auto_commit)
      props.put(kafka::EXCLUDE_INTERNAL_TOPICS_CONFIG, exclude_internal_topics) unless exclude_internal_topics.nil?
      props.put(kafka::FETCH_MAX_WAIT_MS_CONFIG, fetch_max_wait_ms) unless fetch_max_wait_ms.nil?
      props.put(kafka::FETCH_MIN_BYTES_CONFIG, fetch_min_bytes) unless fetch_min_bytes.nil?
      props.put(kafka::GROUP_ID_CONFIG, group_id)
      props.put(kafka::HEARTBEAT_INTERVAL_MS_CONFIG, heartbeat_interval_ms) unless heartbeat_interval_ms.nil?
      props.put(kafka::KEY_DESERIALIZER_CLASS_CONFIG, key_deserializer_class)
      props.put(kafka::MAX_PARTITION_FETCH_BYTES_CONFIG, max_partition_fetch_bytes) unless max_partition_fetch_bytes.nil?
      props.put(kafka::MAX_POLL_RECORDS_CONFIG, max_poll_records) unless max_poll_records.nil?
      props.put(kafka::METADATA_MAX_AGE_MS_CONFIG, metadata_max_age_ms) unless metadata_max_age_ms.nil?
      props.put(kafka::PARTITION_ASSIGNMENT_STRATEGY_CONFIG, partition_assignment_strategy) unless partition_assignment_strategy.nil?
      props.put(kafka::RECEIVE_BUFFER_CONFIG, receive_buffer_bytes) unless receive_buffer_bytes.nil?
      props.put(kafka::RECONNECT_BACKOFF_MS_CONFIG, reconnect_backoff_ms) unless reconnect_backoff_ms.nil?
      props.put(kafka::REQUEST_TIMEOUT_MS_CONFIG, request_timeout_ms) unless request_timeout_ms.nil?
      props.put(kafka::RETRY_BACKOFF_MS_CONFIG, retry_backoff_ms) unless retry_backoff_ms.nil?
      props.put(kafka::SEND_BUFFER_CONFIG, send_buffer_bytes) unless send_buffer_bytes.nil?
      props.put(kafka::SESSION_TIMEOUT_MS_CONFIG, session_timeout_ms) unless session_timeout_ms.nil?
      props.put(kafka::VALUE_DESERIALIZER_CLASS_CONFIG, value_deserializer_class)

      props.put("security.protocol", 'SASL_SSL')
      props.put("sasl.mechanism", 'PLAIN')
      props.put("ssl.protocol", "TLSv1.2")
      props.put("ssl.enabled.protocols", "TLSv1.2")
      props.put("ssl.endpoint.identification.algorithm", "HTTPS")
      props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"****\" password=\"****\";")

      # if security_protocol == "SSL"
      #   set_trustore_keystore_config(props)
      # elsif security_protocol == "SASL_PLAINTEXT"
      #   set_sasl_config(props)
      # elsif security_protocol == "SASL_SSL"
      #   set_trustore_keystore_config(props)
      #   set_sasl_config(props)
      # end

      org.apache.kafka.clients.consumer.KafkaConsumer.new(props)
    rescue => e
      logger.error("Unable to create Kafka consumer from given configuration",
                   :kafka_error_message => e,
                   :cause => e.respond_to?(:getCause) ? e.getCause() : nil)
      throw e
    end
  end

最后更新我的输入配置：

input {
    kafka {
        bootstrap_servers =>  "kafka04-prod01.messagehub.services.eu-de.bluemix.net:9093,kafka05-prod01.messagehub.services.eu-de.bluemix.net:9093,kafka01-prod01.messagehub.services.eu-de.bluemix.net:9093,kafka03-prod01.messagehub.services.eu-de.bluemix.net:9093,kafka02-prod01.messagehub.services.eu-de.bluemix.net:9093"
        topics => [
                "transactions_load"
            ]
    }
}
于 2017-10-09T07:10:50.437 回答