You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by or...@apache.org on 2022/08/08 15:43:57 UTC
[camel] branch main updated (e884743e501 -> 28c7d570791)
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
from e884743e501 CAMEL-18317: camel-salesforce: Add `subscribe` operation.
new 4deb01d4a15 (chores) camel-kafka: fixed typo in the resume strategy
new 4b9374d1865 (chores) camel-kafka: add log guard on resume strategy critical path
new 28c7d570791 CAMEL-18362: rework the consumer/producer
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../kafka/SingleNodeKafkaResumeStrategy.java | 200 ++++++++++-----------
1 file changed, 91 insertions(+), 109 deletions(-)
[camel] 02/03: (chores) camel-kafka: add log guard on resume strategy critical path
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 4b9374d1865034ee3e1a6677707dbb52c2535239
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Mon Aug 8 16:16:37 2022 +0200
(chores) camel-kafka: add log guard on resume strategy critical path
---
.../processor/resume/kafka/SingleNodeKafkaResumeStrategy.java | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java b/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
index 007e9d29bb7..b731e9b0c1c 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
@@ -130,7 +130,9 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
OffsetKey<?> key = offset.getOffsetKey();
Offset<?> offsetValue = offset.getLastOffset();
- LOG.debug("Updating offset on Kafka with key {} to {}", key.getValue(), offsetValue.getValue());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Updating offset on Kafka with key {} to {}", key.getValue(), offsetValue.getValue());
+ }
ByteBuffer keyBuffer = key.serialize();
ByteBuffer valueBuffer = offsetValue.serialize();
@@ -173,7 +175,9 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
for (ConsumerRecord<byte[], byte[]> record : records) {
byte[] value = record.value();
- LOG.trace("Read from Kafka: {}", value);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Read from Kafka: {}", value);
+ }
if (!deserializable.deserialize(ByteBuffer.wrap(record.key()), ByteBuffer.wrap(record.value()))) {
LOG.warn("Deserializer indicates that this is the last record to deserialize");
[camel] 03/03: CAMEL-18362: rework the consumer/producer
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 28c7d570791a478ef56ed728bed8d6c300a466af
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Mon Aug 8 16:31:58 2022 +0200
CAMEL-18362: rework the consumer/producer
- Ensure the consumer is mostly isolated in its own thread
- Prevent closing the producer while an offset update is happening
---
.../kafka/SingleNodeKafkaResumeStrategy.java | 190 +++++++++------------
1 file changed, 84 insertions(+), 106 deletions(-)
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java b/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
index b731e9b0c1c..a527b66c9d7 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
@@ -23,13 +23,13 @@ import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
import java.util.Objects;
-import java.util.Properties;
import java.util.Queue;
-import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.resume.Cacheable;
@@ -41,7 +41,6 @@ import org.apache.camel.resume.ResumeAdapter;
import org.apache.camel.resume.cache.ResumeCache;
import org.apache.camel.util.IOHelper;
import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -50,6 +49,7 @@ import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.WakeupException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -70,12 +70,12 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
private ResumeAdapter adapter;
private final KafkaResumeStrategyConfiguration resumeStrategyConfiguration;
private final ExecutorService executorService;
+ private final ReentrantLock lock = new ReentrantLock();
/**
* Builds an instance of this class
*
* @param resumeStrategyConfiguration the configuration to use for this strategy instance
- *
*/
public SingleNodeKafkaResumeStrategy(KafkaResumeStrategyConfiguration resumeStrategyConfiguration) {
this.resumeStrategyConfiguration = resumeStrategyConfiguration;
@@ -86,7 +86,6 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
* Builds an instance of this class
*
* @param resumeStrategyConfiguration the configuration to use for this strategy instance
- *
*/
public SingleNodeKafkaResumeStrategy(KafkaResumeStrategyConfiguration resumeStrategyConfiguration,
ExecutorService executorService) {
@@ -98,11 +97,10 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
* Sends data to a topic. The records will always be sent asynchronously. If there's an error, a producer error
* counter will be increased.
*
- * @see SingleNodeKafkaResumeStrategy#getProducerErrors()
* @param message the message to send
* @throws ExecutionException
* @throws InterruptedException
- *
+ * @see SingleNodeKafkaResumeStrategy#getProducerErrors()
*/
protected void produce(byte[] key, byte[] message) throws ExecutionException, InterruptedException {
ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(resumeStrategyConfiguration.getTopic(), key, message);
@@ -125,8 +123,6 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
@Override
public void updateLastOffset(T offset) throws Exception {
- createProducer();
-
OffsetKey<?> key = offset.getOffsetKey();
Offset<?> offsetValue = offset.getLastOffset();
@@ -137,39 +133,63 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
ByteBuffer keyBuffer = key.serialize();
ByteBuffer valueBuffer = offsetValue.serialize();
- produce(keyBuffer.array(), valueBuffer.array());
+ try {
+ lock.lock();
+ produce(keyBuffer.array(), valueBuffer.array());
+ } finally {
+ lock.unlock();
+ }
doAdd(key, offsetValue);
}
/**
* Loads the existing data into the cache
- *
- * @throws Exception
*/
- public void loadCache() throws Exception {
- createConsumer();
-
- subscribe();
-
- LOG.debug("Loading records from topic {}", resumeStrategyConfiguration.getTopic());
-
+ @Override
+ public void loadCache() {
if (!(adapter instanceof Deserializable)) {
throw new RuntimeCamelException("Cannot load data for an adapter that is not deserializable");
}
- executorService.submit(() -> refresh());
+ executorService.submit(this::refresh);
+ }
+
+ /**
+ * Launch a thread to refresh the offsets periodically
+ */
+ private void refresh() {
+ LOG.trace("Creating a offset cache refresher");
+
+ try {
+ consumer = createConsumer();
+
+ subscribe(consumer);
+
+ LOG.debug("Loading records from topic {}", resumeStrategyConfiguration.getTopic());
+ consumer.subscribe(Collections.singletonList(getResumeStrategyConfiguration().getTopic()));
+
+ poll(consumer);
+ } catch (WakeupException e) {
+ LOG.info("Kafka consumer was interrupted during a blocking call");
+ } catch (Exception e) {
+ LOG.error("Error while refreshing the local cache: {}", e.getMessage(), e);
+ } finally {
+ if (consumer != null) {
+ consumer.unsubscribe();
+ consumer.close(Duration.ofSeconds(5));
+ }
+ }
}
- protected void poll() {
+ protected void poll(Consumer<byte[], byte[]> consumer) {
Deserializable deserializable = (Deserializable) adapter;
- ConsumerRecords<byte[], byte[]> records;
do {
- records = consume();
+ ConsumerRecords<byte[], byte[]> records = consume(consumer);
if (records.isEmpty()) {
- break;
+ continue;
}
for (ConsumerRecord<byte[], byte[]> record : records) {
@@ -191,7 +211,7 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
*
* @param topic the topic to consume the messages from
*/
- protected void checkAndSubscribe(String topic) {
+ protected void checkAndSubscribe(Consumer<byte[], byte[]> consumer, String topic) {
if (!subscribed) {
consumer.subscribe(Collections.singletonList(topic));
@@ -205,26 +225,18 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
* @param topic the topic to consume the messages from
* @param remaining the number of messages to rewind from the last offset position (used to fill the cache)
*/
- public void checkAndSubscribe(String topic, long remaining) {
+ public void checkAndSubscribe(Consumer<byte[], byte[]> consumer, String topic, long remaining) {
if (!subscribed) {
- consumer.subscribe(Collections.singletonList(topic), getConsumerRebalanceListener(remaining));
-
+ consumer.subscribe(Collections.singletonList(topic), getConsumerRebalanceListener(consumer, remaining));
subscribed = true;
}
}
- /**
- * Creates a new consumer rebalance listener. This can be useful for setting the exact Kafka offset when necessary
- * to read a limited amount of messages or customize the resume strategy behavior when a rebalance occurs.
- *
- * @param remaining the number of remaining messages on the topic to try to collect
- * @return
- */
- protected ConsumerRebalanceListener getConsumerRebalanceListener(long remaining) {
+ private ConsumerRebalanceListener getConsumerRebalanceListener(Consumer<byte[], byte[]> consumer, long remaining) {
return new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> collection) {
-
+ // NO-OP
}
@Override
@@ -244,45 +256,15 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
};
}
- /**
- * Unsubscribe from the topic
- */
- protected void unsubscribe() {
- try {
- consumer.unsubscribe();
- } catch (IllegalStateException e) {
- LOG.warn("The consumer is likely already closed. Skipping unsubscribing from {}",
- resumeStrategyConfiguration.getTopic());
- } catch (Exception e) {
- LOG.error("Error unsubscribing from the Kafka topic {}: {}", resumeStrategyConfiguration.getTopic(), e.getMessage(),
- e);
- }
- }
-
/**
* Consumes message from the topic previously setup
*
* @return An instance of the consumer records
*/
- protected ConsumerRecords<byte[], byte[]> consume() {
- int retries = 10;
-
- return consume(retries);
- }
-
- /**
- * Consumes message from the topic previously setup
- *
- * @param retries how many times to retry consuming data from the topic
- * @return An instance of the consumer records
- */
- protected ConsumerRecords<byte[], byte[]> consume(int retries) {
- while (retries > 0) {
- ConsumerRecords<byte[], byte[]> records = consumer.poll(pollDuration);
- if (!records.isEmpty()) {
- return records;
- }
- retries--;
+ protected ConsumerRecords<byte[], byte[]> consume(Consumer<byte[], byte[]> consumer) {
+ ConsumerRecords<byte[], byte[]> records = consumer.poll(pollDuration);
+ if (!records.isEmpty()) {
+ return records;
}
return ConsumerRecords.empty();
@@ -307,17 +289,17 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
return ConsumerRecords.empty();
}
- public void subscribe() throws Exception {
+ private void subscribe(Consumer<byte[], byte[]> consumer) {
if (adapter instanceof Cacheable) {
ResumeCache<?> cache = ((Cacheable) adapter).getCache();
if (cache.capacity() >= 1) {
- checkAndSubscribe(resumeStrategyConfiguration.getTopic(), cache.capacity());
+ checkAndSubscribe(consumer, resumeStrategyConfiguration.getTopic(), cache.capacity());
} else {
- checkAndSubscribe(resumeStrategyConfiguration.getTopic());
+ checkAndSubscribe(consumer, resumeStrategyConfiguration.getTopic());
}
} else {
- checkAndSubscribe(resumeStrategyConfiguration.getTopic());
+ checkAndSubscribe(consumer, resumeStrategyConfiguration.getTopic());
}
}
@@ -356,21 +338,38 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
}
}
- private void createConsumer() {
- if (consumer == null) {
- consumer = new KafkaConsumer<>(resumeStrategyConfiguration.getConsumerProperties());
- }
+ private Consumer<byte[], byte[]> createConsumer() {
+ return new KafkaConsumer<>(resumeStrategyConfiguration.getConsumerProperties());
}
@Override
public void stop() {
- LOG.info("Closing the Kafka producer");
- IOHelper.close(producer, "Kafka producer", LOG);
+ try {
+ LOG.trace("Trying to obtain a lock for closing the producer");
+ if (!lock.tryLock(1, TimeUnit.SECONDS)) {
+ LOG.warn("Failed to obtain a lock for closing the producer. Force closing the producer ...");
+ }
- LOG.info("Closing the Kafka consumer");
- IOHelper.close(producer, "Kafka consumer", LOG);
+ LOG.info("Closing the Kafka producer");
+ IOHelper.close(producer, "Kafka producer", LOG);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ lock.unlock();
+ }
- executorService.shutdown();
+ try {
+ LOG.info("Closing the Kafka consumer");
+ consumer.wakeup();
+ executorService.shutdown();
+
+ if (!executorService.awaitTermination(2, TimeUnit.SECONDS)) {
+ LOG.warn("Kafka consumer did not shutdown within 2 seconds");
+ executorService.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
}
@Override
@@ -381,6 +380,7 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
@Override
public void start() {
LOG.info("Starting the kafka resume strategy");
+ createProducer();
}
public Duration getPollDuration() {
@@ -391,10 +391,6 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
this.pollDuration = Objects.requireNonNull(pollDuration, "The poll duration cannot be null");
}
- protected Consumer<byte[], byte[]> getConsumer() {
- return consumer;
- }
-
protected Producer<byte[], byte[]> getProducer() {
return producer;
}
@@ -410,22 +406,4 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
return resumeStrategyConfiguration;
}
- /**
- * Launch a thread to refresh the offsets periodically
- */
- private void refresh() {
- LOG.trace("Creating a offset cache refresher");
- try {
- Properties prop = (Properties) getResumeStrategyConfiguration().getConsumerProperties().clone();
- prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
-
- try (Consumer<byte[], byte[]> consumer = new KafkaConsumer<>(prop)) {
- consumer.subscribe(Collections.singletonList(getResumeStrategyConfiguration().getTopic()));
-
- poll();
- }
- } catch (Exception e) {
- LOG.error("Error while refreshing the local cache: {}", e.getMessage(), e);
- }
- }
}
[camel] 01/03: (chores) camel-kafka: fixed typo in the resume strategy
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 4deb01d4a15dad784b5e7870119c485c7265050a
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Mon Aug 8 11:20:23 2022 +0200
(chores) camel-kafka: fixed typo in the resume strategy
---
.../camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java b/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
index db87f65074a..007e9d29bb7 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
@@ -176,7 +176,7 @@ public class SingleNodeKafkaResumeStrategy<T extends Resumable> implements Kafka
LOG.trace("Read from Kafka: {}", value);
if (!deserializable.deserialize(ByteBuffer.wrap(record.key()), ByteBuffer.wrap(record.value()))) {
- LOG.warn("Deserializar indicates that this is the last record to deserialize");
+ LOG.warn("Deserializer indicates that this is the last record to deserialize");
}
}
} while (true);