streamnative / logstash-input-pulsar

Logstash Input Pulsar Plugin

This is a Java plugin for Logstash.

It is fully free and fully open source. The license is Apache 2.0, meaning you are free to use it however you want.

This input will read events from a Pulsar topic.

This plugin uses Pulsar Client 2.10.2. For broker compatibility, see the official Pulsar compatibility reference. If the compatibility wiki is not up-to-date, please contact Pulsar support/community to confirm compatibility.

If you require features not yet available in this plugin (including client version upgrades), please file an issue with details about what you need.

Pulsar Input Configuration Options

This plugin supports these configuration options.

Settings	Input type	Required
serviceUrl	string	Yes
topics	array	Yes
subscriptionName	string	Yes
consumerName	string	Yes
subscriptionType	string, one of["Shared","Exclusive","Failover","Key_shared"]	No
subscriptionInitialPosition	string, one of["Latest","Earliest"]	No
codec	string, one of ["plain","json"]	No
protocols	array, ciphers list. default is TLSv1.2	No
ciphers	array,ciphers list	No
enable_tls	boolean, one of [true, false]. default is false	No
tls_trust_store_path	string, required if enable_tls is set to true	No
tls_trust_store_password	string, default is empty	No
enable_tls_hostname_verification	boolean, one of [true, false]. default is false	No
allow_tls_insecure_connection	boolean, one of [true, false].default is false	No
auth_plugin_class_name	string	No
auth_params	string	No

Example

pulsar without tls & token

input{
  pulsar{
    serviceUrl => "pulsar://127.0.0.1:6650"
    codec => "json"
    topics => [ 
        "persistent://public/default/topic1", 
        "persistent://public/default/topic2"
    ]
    subscriptionName => "my_consumer"
    subscriptionType => "Shared"
    subscriptionInitialPosition => "Earliest"
  }
}

pulsar with token

input{
  pulsar{
    serviceUrl => "pulsar://127.0.0.1:6650"
    codec => "plain"
    topics => [ 
        "persistent://public/default/topic1", 
        "persistent://public/default/topic2"
    ]
    subscriptionName => "my_subscription"
    subscriptionType => "Shared"
    subscriptionInitialPosition => "Earliest"
    auth_plugin_class_name => "org.apache.pulsar.client.impl.auth.AuthenticationToken"
    auth_params => "token:${token}"
  }
}

Installation

1. Get the latest zip file from release page. https://github.com/streamnative/logstash-input-pulsar/releases

2. Install this plugin using logstash preoffline command.

bin/logstash-plugin install file://{PATH_TO}/logstash-input-pulsar-2.7.1.zip

logstash-input-pulsar's Issues

pulsar threads did not consume message from pulsar

we use logstash-input-pulsar to consume messages fron pulsar . After running for some time, we found that logstash-input-pulsar thread did not consume messages fron pulsar. we hope that pulsar can solve this problem

add support for avro

request support of codec for AVRO schema.

ERROR: Unknown setting 'auth_params' specified for plugin 'pulsar'

[2022-12-06T16:26:23,874][INFO ][org.reflections.Reflections] Reflections took 185 ms to scan 1 urls, producing 125 keys and 438 values
[2022-12-06T16:26:24,015][ERROR][co.elastic.logstash.api.PluginHelper] Unknown setting 'auth_params' specified for plugin 'pulsar'
[2022-12-06T16:26:24,016][ERROR][co.elastic.logstash.api.PluginHelper] Unknown setting 'auth_plugin_class_name' specified for plugin 'pulsar'
[2022-12-06T16:26:24,020][ERROR][logstash.agent           ] Failed to execute action {:action=>LogStash::PipelineAction::Create/pipeline_id:main, :exception=>"Java::JavaLang::IllegalStateException", :message=>"Unable to configure plugins: Config errors found for plugin 'pulsar'", :backtrace=>["org.logstash.config.ir.CompiledPipeline.<init>(CompiledPipeline.java:120)", "org.logstash.execution.JavaBasePipelineExt.initialize(JavaBasePipelineExt.java:85)", "org.logstash.execution.JavaBasePipelineExt$INVOKER$i$1$0$initialize.call(JavaBasePipelineExt$INVOKER$i$1$0$initialize.gen)", "org.jruby.internal.runtime.methods.JavaMethod$JavaMethodN.call(JavaMethod.java:846)", "org.jruby.ir.runtime.IRRuntimeHelpers.instanceSuper(IRRuntimeHelpers.java:1229)", "org.jruby.ir.runtime.IRRuntimeHelpers.instanceSuperSplatArgs(IRRuntimeHelpers.java:1202)", "org.jruby.ir.targets.indy.InstanceSuperInvokeSite.invoke(InstanceSuperInvokeSite.java:29)", "opt.elastic.logstash_minus_8_dot_5_dot_2.logstash_minus_core.lib.logstash.java_pipeline.RUBY$method$initialize$0(/opt/elastic/logstash-8.5.2/logstash-core/lib/logstash/java_pipeline.rb:48)", "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:139)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:112)", "org.jruby.runtime.callsite.CachingCallSite.cacheAndCall(CachingCallSite.java:329)", "org.jruby.runtime.callsite.CachingCallSite.call(CachingCallSite.java:87)", "org.jruby.RubyClass.newInstance(RubyClass.java:911)", "org.jruby.RubyClass$INVOKER$i$newInstance.call(RubyClass$INVOKER$i$newInstance.gen)", "org.jruby.ir.targets.indy.InvokeSite.invoke(InvokeSite.java:208)", "opt.elastic.logstash_minus_8_dot_5_dot_2.logstash_minus_core.lib.logstash.pipeline_action.create.RUBY$method$execute$0(/opt/elastic/logstash-8.5.2/logstash-core/lib/logstash/pipeline_action/create.rb:50)", "opt.elastic.logstash_minus_8_dot_5_dot_2.logstash_minus_core.lib.logstash.pipeline_action.create.RUBY$method$execute$0$__VARARGS__(/opt/elastic/logstash-8.5.2/logstash-core/lib/logstash/pipeline_action/create.rb:49)", "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:139)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:112)", "org.jruby.ir.targets.indy.InvokeSite.invoke(InvokeSite.java:208)", "opt.elastic.logstash_minus_8_dot_5_dot_2.logstash_minus_core.lib.logstash.agent.RUBY$block$converge_state$2(/opt/elastic/logstash-8.5.2/logstash-core/lib/logstash/agent.rb:386)", "org.jruby.runtime.CompiledIRBlockBody.callDirect(CompiledIRBlockBody.java:141)", "org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:64)", "org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:58)", "org.jruby.runtime.Block.call(Block.java:143)", "org.jruby.RubyProc.call(RubyProc.java:309)", "org.jruby.internal.runtime.RubyRunnable.run(RubyRunnable.java:107)", "java.base/java.lang.Thread.run(Thread.java:833)"]}
[2022-12-06T16:26:24,134][INFO ][logstash.runner          ] Logstash shut down.
[2022-12-06T16:26:24,147][FATAL][org.logstash.Logstash    ] Logstash stopped processing because of an error: (SystemExit) exit
org.jruby.exceptions.SystemExit: (SystemExit) exit
	at org.jruby.RubyKernel.exit(org/jruby/RubyKernel.java:790) ~[jruby.jar:?]
	at org.jruby.RubyKernel.exit(org/jruby/RubyKernel.java:753) ~[jruby.jar:?]
	at opt.elastic.logstash_minus_8_dot_5_dot_2.lib.bootstrap.environment.<main>(/opt/elastic/logstash-8.5.2/lib/bootstrap/environment.rb:91) ~[?:?]

Is this pulgin will support codec avro？

Python command 'input' not working.

Hello guys! I'm learning python and studying, i started by now to use Pulsar in alternative to VS code, because i can't use it anymore due my mac version.
I'm trying a simple code, for example: input ("yes or no")

I've tried to put de UTF-8 encoder, but it didn't work. Can someone help me?

New release

Hi whoever this may concern,

Last time there was a release was Jun 26 2021. https://github.com/streamnative/logstash-input-pulsar/releases/tag/2.7.1

This version is missing several fields that are quite useful. The only way for my colleague and I to actually use this currently is to build from src. This plugin works like a charm, so thank you for having it open source.

Please create a new release as soon as possible.

Have a lovely day. Ping me if you need anything from me.