Skip to content
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

Jump to bottom

[7.67.x-blue][RHPAM-4898] Fix concurrency issue in KafkaServerRegistration #3042

Merged
merged 4 commits into from
Mar 21, 2024
Merged

[7.67.x-blue][RHPAM-4898] Fix concurrency issue in KafkaServerRegistration #3042

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
253400e
[FIX_KAFKA_EXECUTOR] fix executor service
elguardian Feb 29, 2024
2382b99
Fix logging output for processed ConsumerRecords
martinweiler Feb 29, 2024
bddc2a6
Avoid synchronization on topic maps
martinweiler Feb 29, 2024
8aefd17
Restore synchronized method signatures
martinweiler Mar 1, 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
10 changes: 3 additions & 7 deletions ...-services-kafka/src/main/java/org/kie/server/services/jbpm/kafka/KafkaServerConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
Expand Down Expand Up @@ -112,10 +113,7 @@ private void registrationUpdated(Set<String> topics2Register, boolean shouldInit
consumer = consumerSupplier.get();
consumer.subscribe(topics2Register);
logger.debug("Created kafka consumer with these topics registered {}", topics2Register);
notifyService.set(
new ThreadPoolExecutor(1, Integer.getInteger(KAFKA_EXTENSION_PREFIX + "maxNotifyThreads", 10),
60L,
TimeUnit.SECONDS, new LinkedBlockingQueue<>()));
notifyService.set(Executors.newFixedThreadPool(Integer.getInteger(KAFKA_EXTENSION_PREFIX + "maxNotifyThreads", 10)));
new Thread(this).start();
}
} else {
Expand Down Expand Up @@ -198,12 +196,10 @@ private void processEvents(ConsumerRecords<String, byte[]> events) {


private void printEventsLog(ConsumerRecords<String, byte[]> events) {
Map<String, Integer> eventsPerTopic = new HashMap<>();
for (ConsumerRecord<String, byte[]> event : events) {
logger.trace("Kafka event received {}", event);
eventsPerTopic.compute(event.topic(), (k, v) -> v == null ? 1 : v++);
}
logger.debug("Number of events received per topic {}", eventsPerTopic);
logger.debug("Number of events received per topic {}", events.count());
}

@FunctionalInterface
Expand Down
7 changes: 4 additions & 3 deletions ...vices-kafka/src/main/java/org/kie/server/services/jbpm/kafka/KafkaServerRegistration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.function.Consumer;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.jbpm.kie.services.impl.KModuleDeploymentUnit;
Expand All @@ -42,8 +43,8 @@

class KafkaServerRegistration {

private Map<String, Map<SignalDesc, Map<DeploymentId, SortedSet<VersionedDeploymentId>>>> topic2Signal = new HashMap<>();
private Map<String, Map<MessageDesc, Map<DeploymentId, SortedSet<VersionedDeploymentId>>>> topic2Message = new HashMap<>();
private Map<String, Map<SignalDesc, Map<DeploymentId, SortedSet<VersionedDeploymentId>>>> topic2Signal = new ConcurrentHashMap<>();
private Map<String, Map<MessageDesc, Map<DeploymentId, SortedSet<VersionedDeploymentId>>>> topic2Message = new ConcurrentHashMap<>();

synchronized void close() {
topic2Signal.clear();
Expand Down Expand Up @@ -100,7 +101,7 @@ void forEachMessage(ConsumerRecord<String, byte[]> event, KafkaServerEventProces
forEach(topic2Message, event, eventProcessor);
}

private synchronized <T extends SignalDescBase> void forEach(Map<String, Map<T, Map<DeploymentId, SortedSet<VersionedDeploymentId>>>> topic2SignalBase,
private <T extends SignalDescBase> void forEach(Map<String, Map<T, Map<DeploymentId, SortedSet<VersionedDeploymentId>>>> topic2SignalBase,
ConsumerRecord<String, byte[]> event,
KafkaServerEventProcessor<T> processor) {
Map<T, Map<DeploymentId, SortedSet<VersionedDeploymentId>>> signalInfo = topic2SignalBase.get(event.topic());
Expand Down
Loading