连接到 IBM Message Hub 的 kafka 输入插件出现不可恢复的错误
Unrecoverable error with kafka input plugin connecting to IBM Message Hub
我正在尝试从 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 example，我缺少以下 kafka 配置属性：
ssl.protocol=TLSv1.2
ssl.enabled.protocols=TLSv1.2
ssl.endpoint.identification.algorithm=HTTPS
但是，我无法从 logstash documentation 中看到如何设置这些。
我的 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）
我的解决方案是：
# 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"
]
}
}
我正在尝试从 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 example，我缺少以下 kafka 配置属性：
ssl.protocol=TLSv1.2
ssl.enabled.protocols=TLSv1.2
ssl.endpoint.identification.algorithm=HTTPS
但是，我无法从 logstash documentation 中看到如何设置这些。
我的 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）
我的解决方案是：
# 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"
]
}
}