NullPointerException when configuring SSL in the wrong way

Version: logstash-input-kafka-8.0.4
Operating System: Any
Config File (if you have sensitive info, please remove it):

     kafka { 
        bootstrap_servers => 'xxxxx:9092' 
        topics => ["xxxxx"] 
        codec => json { } 
        security_protocol => "SSL" 
        ssl_keystore_location => "/path/to/xxxx.jks" 
        ssl_keystore_password => "xxx"
    }

Sample Data: None
Steps to Reproduce:

Start logstash.

Inspect logs:

[2018-04-05T10:58:08,080][ERROR][logstash.pipeline        ] A plugin had an unrecoverable error. Will restart this plugin.
  Pipeline_id:main
  Plugin: <LogStash::Inputs::Kafka ......config here>
  Error: uncaught throw java.lang.NullPointerException
  Exception: UncaughtThrowError
  Stack: org/jruby/RubyKernel.java:1137:in `throw'
/Users/Gabriel/Documents/ElasticSearch/logstash-6.2.3/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.0.4/lib/logstash/inputs/kafka.rb:329:in `create_consumer'
/Users/Gabriel/Documents/ElasticSearch/logstash-6.2.3/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.0.4/lib/logstash/inputs/kafka.rb:224:in `block in run'
org/jruby/RubyFixnum.java:305:in `times'
org/jruby/RubyEnumerator.java:323:in `each'
org/jruby/RubyEnumerable.java:830:in `map'
/Users/Gabriel/Documents/ElasticSearch/logstash-6.2.3/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.0.4/lib/logstash/inputs/kafka.rb:224:in `run'
/Users/Gabriel/Documents/ElasticSearch/logstash-6.2.3/logstash-core/lib/logstash/pipeline.rb:516:in `inputworker'
/Users/Gabriel/Documents/ElasticSearch/logstash-6.2.3/logstash-core/lib/logstash/pipeline.rb:509:in `block in start_input'

It seems that the following lines are causing this: https://github.com/logstash-plugins/logstash-input-kafka/blob/v8.0.4/lib/logstash/inputs/kafka.rb#L333-L343

I added some debugging and it seems that when security_protocol => "SSL" then the ssl_truststore_location is required. But, the error message isn't clear enough for all these.

Once you add the trustore location, then we will bypass this exception. But, if the trustore is invalid then you get a new UncaughtThrowError:

[2018-04-05T11:02:30,816][ERROR][logstash.pipeline        ] A plugin had an unrecoverable error. Will restart this plugin.
  Pipeline_id:main
  Plugin: <LogStash::Inputs::Kafka configuration here poll_timeout_ms=>100, sasl_mechanism=>"GSSAPI", decorate_events=>false>
  Error: uncaught throw org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
  Exception: UncaughtThrowError
  Stack: org/jruby/RubyKernel.java:1137:in `throw'
/Users/Gabriel/Documents/ElasticSearch/logstash-6.2.3/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.0.4/lib/logstash/inputs/kafka.rb:329:in `create_consumer'
/Users/Gabriel/Documents/ElasticSearch/logstash-6.2.3/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.0.4/lib/logstash/inputs/kafka.rb:224:in `block in run'
org/jruby/RubyFixnum.java:305:in `times'
org/jruby/RubyEnumerator.java:323:in `each'
org/jruby/RubyEnumerable.java:830:in `map'
/Users/Gabriel/Documents/ElasticSearch/logstash-6.2.3/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.0.4/lib/logstash/inputs/kafka.rb:224:in `run'
/Users/Gabriel/Documents/ElasticSearch/logstash-6.2.3/logstash-core/lib/logstash/pipeline.rb:516:in `inputworker'
/Users/Gabriel/Documents/ElasticSearch/logstash-6.2.3/logstash-core/lib/logstash/pipeline.rb:509:in `block in start_input'

So i think here we need a couple of things:

To enhance the error handling to better understand what is wrong.
Enhance the documentation to explain what is required and what not to build the trustore/keystore settings (and possibly add examples?).

logstash-plugins / logstash-input-kafka

NullPointerException when configuring SSL in the wrong way #258