Skip to content

Add async as config option #31

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Add async as config option #31

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ac44712
Add changes
Andyandpandy Apr 13, 2023
ef090c8
Update src/main/java/org/apache/pulsar/logstash/outputs/Pulsar.java
Andyandpandy Jun 8, 2023
7b2f968
Update src/main/java/org/apache/pulsar/logstash/outputs/Pulsar.java
Andyandpandy Jun 8, 2023
882ff02
Update README.md
Andyandpandy Jun 8, 2023
7c1bc75
Update README.md
Andyandpandy Jun 8, 2023
2fb1f80
Update send function to also include s as param to avoid runtime errors
Andyandpandy Jun 8, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,15 @@ This plugin supports these configuration options.
| ciphers | array, ciphers list | No |
| enable_token | boolean, one of [true, false]. default is false | No |
| auth_plugin_params_String | string | No |
| enable_async | boolean, one of [true, false]. default is false | No |


## Sync vs Async

Sync is slower because it requires verification that a message is received. Sync supports exactly/effectively once messaging. This means there is a big network latency increase here.

Async is faster because it sends messages out and doesn't care if the message was received/processed or not. Async should only be used if you want "at most once" messaging and don't care if messages are lost.

# Example
pulsar without tls & token
```
Expand Down
23 changes: 20 additions & 3 deletions src/main/java/org/apache/pulsar/logstash/outputs/Pulsar.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,9 @@ public class Pulsar implements Output {
private static final PluginConfigSpec<String> CONFIG_AUTH_PLUGIN_PARAMS_STRING =
PluginConfigSpec.stringSetting("auth_plugin_params_String","");

private static final PluginConfigSpec<Boolean> CONFIG_ENABLE_ASYNC =
PluginConfigSpec.booleanSetting("enable_async",false);

private final CountDownLatch done = new CountDownLatch(1);

private final String producerName;
Expand Down Expand Up @@ -125,6 +128,9 @@ public class Pulsar implements Output {
//Token
private final boolean enableToken;

// Sends messages async if set, otherwise sync
private final boolean enableAsync;

// TODO: batchingMaxPublishDelay milliseconds

// TODO: sendTimeoutMs milliseconds 30000
Expand All @@ -148,6 +154,7 @@ public Pulsar(final String id, final Configuration configuration, final Context

enableTls = configuration.get(CONFIG_ENABLE_TLS);
enableToken = configuration.get(CONFIG_ENABLE_TOKEN);
enableAsync = configuration.get(CONFIG_ENABLE_ASYNC);

try {
if(enableTls && enableToken){
Expand Down Expand Up @@ -222,15 +229,25 @@ public void output(final Collection<Event> events) {
codec.encode(event, baos);
String s = baos.toString();
logger.debug("topic is {}, message is {}", eventTopic, s);
getProducer(eventTopic).newMessage()
.value(s.getBytes())
.sendAsync();
send(eventTopic, s);
} catch (Exception e) {
logger.error("fail to send message", e);
}
}
}

private void send(String topic, String s){
if (enableAsync){
getProducer(topic).newMessage()
.value(s.getBytes())
.sendAsync();
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We still need to add a callback to print a log when failing to publish messages.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This send function is only called inside a try/catch, where it logs "fail to send message" with the exception. See line 225-235.

Copy link
Member

@RobertIndie RobertIndie Jun 21, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It won't catch the exception thrown from sendAsync. We need to use CompletableFuture.exceptionally to catch it. Please see: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html#exceptionally-java.util.function.Function-

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay.. Am I reading this right?

CompletableFuture.exceptionally description:

Returns a new CompletableFuture that is completed when this CompletableFuture completes, with the result of the given function of the exception triggering this CompletableFuture's completion when it completes exceptionally; otherwise, if this CompletableFuture completes normally, then the returned CompletableFuture also completes normally with the same value.

This doesn't throw a custom Exception that inherits from the class Exception as usual? I don't think I understand what this is describing. I'm trying to decipher it, but does it mean that it returns the function that triggered an Exception rather than actually throwing the Exception. In that case I guess I couldn't try/catch? Or does it mean that I am supposed to just provide a function that is throwable, so when CompletableFuture throws, it triggers my function with where it failed?

Copy link
Member

@RobertIndie RobertIndie Jul 5, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry for the late reply.

The sendAsync will not throw any exceptions. The producer already catches all the exceptions and places them into the Future. You could check the code here.

Therefore if there are any exceptions thrown in the send operation, the function fn in the exceptionally will be called. And I think we need to print the error log in that function.

Or does it mean that I am supposed to just provide a function that is throwable, so when CompletableFuture throws, it triggers my function with where it failed?

That's correct.

However, you still need to use try/catch here, because we couldn't guarantee that this code won't throw the exception:

getProducer(topic).newMessage()
                .value(s.getBytes())
                .sendAsync();

}else {
getProducer(topic).newMessage()
.value(s.getBytes())
.send();
}
}

private org.apache.pulsar.client.api.Producer<byte[]> getProducer(String topic) throws PulsarClientException {

if(producerMap.containsKey(topic)){
Expand Down