Given my tests, "schema_registry_proxy" option does not work on kafka input plugin.

I mean:

• if I set a true URL (ex: "http://myhost:1234"), then I get this error at Logstash startup:

Pipeline error {:pipeline_id=>"kafka", :exception=>#<NoMethodError: undefined method `empty?' for //host.fr:1234:LogStash::Util::SafeURI>, :backtrace=>["C:/Developpement/elastic-stack/products/logstash-7.17.8/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.9.0-java/lib/logstash/plugin_mixins/common.rb:85:in `split_proxy_into_host_and_port'", "C:/Developpement/elastic-stack/products/logstash-7.17.8/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.9.0-java/lib/logstash/plugin_mixins/common.rb:34:in `check_schema_registry_parameters'", "C:/Developpement/elastic-stack/products/logstash-7.17.8/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.9.0-java/lib/logstash/inputs/kafka.rb:271:in `register'"}

It is because URL is parsed as LogStash::Util::SafeURI and this class has no empty? method.

• if I set "" (documentation says that it is possible), then I get this error at Logstash startup:

Failed to execute action {:action=>LogStash::PipelineAction::Create/pipeline_id:kafka, :exception=>"Java::JavaLang::IllegalStateException", :message=>"Unable to configure plugins: (ArgumentError) URI is not valid - host is not specified"

I guess this is because LogStash::Util::SafeURI does not support an empty string as value.

Logstash information:

Please include the following information:

1. Logstash version : 7.17.8
2. Logstash installation source : expanded from tar or zip archive
3. How is Logstash being run : Via command line
4. How was the Logstash Plugin installed : default version provided by Logstash 7.17.8 (10.9.0)

JVM (e.g. java -version): OpenJDK 11
2. JVM installation source : AdoptOpenJDK distribution
3. Value of the JAVA_HOME environment variable if set: C:\Developpement\OpenJDK11_x64

OS version (uname -a if on a Unix-like system): Windows 10

Description of the problem including expected versus actual behavior:

Steps to reproduce:

1. Define kafka input with "schema_registry_key":

kafka {
...
schema_registry_proxy => "http://myhost:1234"
}

2. Start Logstash
3. The error happens during startup

Provide logs (if relevant):
Already provided above.