Skip to content

name single threads of kafka live event listener #97

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

Merged
merged 2 commits into from
Apr 24, 2024
Merged

name single threads of kafka live event listener #97

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
965510c
name single threads of kafka live event listener
Apr 24, 2024
0d5b2b0
update
Apr 24, 2024
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
19 changes: 18 additions & 1 deletion ...stener/src/main/java/org/hypertrace/core/kafka/event/listener/KafkaLiveEventListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import org.apache.kafka.clients.consumer.Consumer;
Expand Down Expand Up @@ -64,7 +65,7 @@ public void close() throws Exception {
public static final class Builder<K, V> {
private final Collection<BiConsumer<? super K, ? super V>> callbacks =
new ConcurrentLinkedQueue<>();
private ExecutorService executorService = Executors.newSingleThreadExecutor();
private ExecutorService executorService;
private boolean cleanupExecutor =
true; // if builder creates executor shutdown executor while closing event listener

Expand All @@ -84,10 +85,26 @@ public Builder<K, V> withExecutorService(

public KafkaLiveEventListener<K, V> build(
String consumerName, Config kafkaConfig, Consumer<K, V> kafkaConsumer) {
if (executorService == null) {
executorService =
Executors.newSingleThreadExecutor(new ListenerThreadFactory(consumerName));
}
return new KafkaLiveEventListener<>(
new KafkaLiveEventListenerCallable<>(consumerName, kafkaConfig, kafkaConsumer, callbacks),
executorService,
cleanupExecutor);
}
}
}

class ListenerThreadFactory implements ThreadFactory {
private final String name;

public ListenerThreadFactory(String consumerName) {
this.name = "kafka-live-event-listener-" + consumerName + "-" + System.currentTimeMillis();
Copy link
Contributor

Choose a reason for hiding this comment

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

Is it common pattern to add time in name?

Copy link
Author

Choose a reason for hiding this comment

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

I added here intentionally to detect if we have multiple threads for same consumer name

Copy link
Author

@kishansairam9 kishansairam9 Apr 24, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

Ideally for one topic there should be only one live event listener hence one thread only. But if it is wrongly used (calling build again and again instead of only once) there might be more than one, which we want to detect and fix

Copy link
Contributor

Choose a reason for hiding this comment

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

If there are 8 StreamThread, there will be 8 Consumers.

If, its more than that, who is caller? How would we fix this? I think, is this called by our application code or by kafka framework itself?

Copy link
Author

Choose a reason for hiding this comment

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

If there are 8 StreamThread, there will be 8 Consumers.

No even if there are 8 stream threads we only want one consumer per topic - singleton instance

Copy link
Author

Choose a reason for hiding this comment

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

If, its more than that, who is caller? How would we fix this? I think, is this called by our application code or by kafka framework itself?

This is used by application code, mainly to listen to change events. If we see more than one thread for a consumer name that means we are rebuilding live event listener under some condition, which is wrong. To fix, you might need to organise the building of live event listener to the place where code is exactly once executed - guice singleton or build topology and pass it down appropriately

Copy link
Contributor

Choose a reason for hiding this comment

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

Suggested change
this.name = "kafka-live-event-listener-" + consumerName + "-" + System.currentTimeMillis();
this.name = "kafka-live-event-listener-" + consumerName + "-" + String.valueOf(System.currentTimeMillis());

Copy link
Author

Choose a reason for hiding this comment

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

image it is not needed

}

public Thread newThread(Runnable r) {
return new Thread(r, name);
}
}
Loading