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/02/07 20:33:24 UTC
[camel] branch main updated: CAMEL-15562: simplify the resume API
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
The following commit(s) were added to refs/heads/main by this push:
new 855f367 CAMEL-15562: simplify the resume API
855f367 is described below
commit 855f367019dda54e6ac83dafab2c78742f0c0717
Author: Otavio Rodolfo Piske <op...@redhat.com>
AuthorDate: Mon Jan 24 14:17:57 2022 +0100
CAMEL-15562: simplify the resume API
Includes:
- adjusted Kafka resume code to the same design as the other components
- adjusted File resume code to the same design as the other components
- adjusted other components to the new API
- offsets management cleanup
- resumables cleanup
- also introduces support for updating the offsets.
---
.../component/aws2/kinesis/Kinesis2Consumer.java | 3 +-
.../kinesis/consumer/KinesisResumeStrategy.java | 1 +
.../KinesisUserConfigurationResumeStrategy.java | 8 +-
.../component/couchdb/CouchDbChangesetTracker.java | 5 +-
.../couchdb/consumer/CouchDbResumable.java | 15 +++-
.../couchdb/consumer/CouchDbResumeStrategy.java | 2 +-
.../LatestUpdateSequenceResumeStrategy.java | 10 ++-
.../apache/camel/component/file/FileConsumer.java | 11 +--
.../apache/camel/component/file/GenericFile.java | 17 ++--
.../camel/component/file/GenericFileConverter.java | 2 +-
.../file/consumer/FileConsumerResumeStrategy.java | 17 +---
.../component/file/consumer/FileResumeSet.java | 26 ++-----
.../file/consumer/FileSetResumeStrategy.java} | 9 +--
.../file/consumer/GenericFileResumable.java} | 9 +--
.../file/consumer/GenericFileResumeStrategy.java} | 9 +--
.../support/KafkaConsumerResumeStrategy.java | 19 +++--
.../kafka/consumer/support/KafkaOffset.java} | 22 ++++--
.../kafka/consumer/support/KafkaResumable.java} | 32 ++++----
.../support/OffsetKafkaConsumerResumeStrategy.java | 18 ++++-
.../support/PartitionAssignmentListener.java | 9 ++-
.../consumer/support/ResumeStrategyFactory.java | 11 ++-
.../SeekPolicyKafkaConsumerResumeStrategy.java | 17 +++-
.../KafkaConsumerWithResumeStrategyIT.java | 14 +++-
.../src/main/java/org/apache/camel/Offset.java | 17 ++--
.../src/main/java/org/apache/camel/Resumable.java | 20 +++--
.../main/java/org/apache/camel/ResumableSet.java | 72 +++++++++++++++++
.../main/java/org/apache/camel/ResumeStrategy.java | 3 +-
...e.java => UpdatableConsumerResumeStrategy.java} | 25 +++---
.../FileConsumerResumeFromOffsetStrategyTest.java | 18 ++---
.../file/FileConsumerResumeStrategyTest.java | 29 +++----
.../main/java/org/apache/camel/resume/Offsets.java | 62 +++++++++++++++
.../java/org/apache/camel/resume/Resumables.java | 91 ++++++++++++++++++++++
32 files changed, 450 insertions(+), 173 deletions(-)
diff --git a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
index 44cec89b..8ad6181 100644
--- a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
+++ b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
@@ -178,7 +178,8 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer {
resumeStrategy = getEndpoint().getConfiguration().getResumeStrategy();
}
- resumeStrategy.resume(req);
+ resumeStrategy.setRequestBuilder(req);
+ resumeStrategy.resume();
}
private Queue<Exchange> createExchanges(List<Record> records) {
diff --git a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/consumer/KinesisResumeStrategy.java b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/consumer/KinesisResumeStrategy.java
index 7e0e5bc..b6a000d 100644
--- a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/consumer/KinesisResumeStrategy.java
+++ b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/consumer/KinesisResumeStrategy.java
@@ -22,4 +22,5 @@ import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest;
public interface KinesisResumeStrategy extends ResumeStrategy<GetShardIteratorRequest.Builder> {
+ void setRequestBuilder(GetShardIteratorRequest.Builder builder);
}
diff --git a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/consumer/KinesisUserConfigurationResumeStrategy.java b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/consumer/KinesisUserConfigurationResumeStrategy.java
index 4d78990..5633443 100644
--- a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/consumer/KinesisUserConfigurationResumeStrategy.java
+++ b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/consumer/KinesisUserConfigurationResumeStrategy.java
@@ -23,6 +23,7 @@ import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
public class KinesisUserConfigurationResumeStrategy implements KinesisResumeStrategy {
private final Kinesis2Configuration configuration;
+ private GetShardIteratorRequest.Builder resumable;
public KinesisUserConfigurationResumeStrategy(Kinesis2Configuration configuration) {
this.configuration = configuration;
@@ -35,7 +36,12 @@ public class KinesisUserConfigurationResumeStrategy implements KinesisResumeStra
}
@Override
- public void resume(GetShardIteratorRequest.Builder resumable) {
+ public void setRequestBuilder(GetShardIteratorRequest.Builder resumable) {
+ this.resumable = resumable;
+ }
+
+ @Override
+ public void resume() {
if (hasSequenceNumber()) {
resumable.startingSequenceNumber(configuration.getSequenceNumber());
}
diff --git a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/CouchDbChangesetTracker.java b/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/CouchDbChangesetTracker.java
index e64d8a1..9e305ad 100644
--- a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/CouchDbChangesetTracker.java
+++ b/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/CouchDbChangesetTracker.java
@@ -55,12 +55,13 @@ public class CouchDbChangesetTracker implements Runnable {
if (sequence == null) {
CouchDbResumeStrategy resumeStrategy = CouchDdResumeStrategyFactory.newResumeStrategy(this.endpoint);
- resumeStrategy.resume(resumable);
+ resumeStrategy.setResumable(resumable);
+ resumeStrategy.resume();
}
LOG.debug("Last sequence [{}]", resumable.getLastOffset());
changes = couchClient.changes().style(endpoint.getStyle()).includeDocs(true)
- .since(resumable.getLastOffset()).heartBeat(endpoint.getHeartbeat()).continuousChanges();
+ .since(resumable.getLastOffset().offset()).heartBeat(endpoint.getHeartbeat()).continuousChanges();
}
@Override
diff --git a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumable.java b/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumable.java
index ac3d679..e6ebd10 100644
--- a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumable.java
+++ b/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumable.java
@@ -17,13 +17,15 @@
package org.apache.camel.component.couchdb.consumer;
+import org.apache.camel.Offset;
import org.apache.camel.Resumable;
import org.apache.camel.component.couchdb.CouchDbClientWrapper;
+import org.apache.camel.resume.Offsets;
/**
* Wraps the resume data for CouchDb
*/
-public class CouchDbResumable implements Resumable<String> {
+public class CouchDbResumable implements Resumable<String, String> {
private final CouchDbClientWrapper clientWrapper;
private String offset;
@@ -33,13 +35,18 @@ public class CouchDbResumable implements Resumable<String> {
}
@Override
- public void setLastOffset(String offset) {
+ public void updateLastOffset(String offset) {
this.offset = offset;
}
@Override
- public String getLastOffset() {
- return offset;
+ public Offset<String> getLastOffset() {
+ return Offsets.of(offset);
+ }
+
+ @Override
+ public String getAddressable() {
+ return null;
}
/**
diff --git a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java b/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java
index cd8ba08..b03b31d 100644
--- a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java
+++ b/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java
@@ -23,5 +23,5 @@ import org.apache.camel.ResumeStrategy;
* Defines a resumable strategy usable by the CouchDB component
*/
public interface CouchDbResumeStrategy extends ResumeStrategy<CouchDbResumable> {
-
+ void setResumable(CouchDbResumable resumable);
}
diff --git a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/LatestUpdateSequenceResumeStrategy.java b/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/LatestUpdateSequenceResumeStrategy.java
index ac80db8..277d842 100644
--- a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/LatestUpdateSequenceResumeStrategy.java
+++ b/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/LatestUpdateSequenceResumeStrategy.java
@@ -23,11 +23,17 @@ import org.apache.camel.component.couchdb.CouchDbClientWrapper;
* A resume strategy that resumes from the last update sequence
*/
public final class LatestUpdateSequenceResumeStrategy implements CouchDbResumeStrategy {
+ private CouchDbResumable resumable;
@Override
- public void resume(CouchDbResumable resumable) {
+ public void setResumable(CouchDbResumable resumable) {
+ this.resumable = resumable;
+ }
+
+ @Override
+ public void resume() {
CouchDbClientWrapper clientWrapper = resumable.getClientWrapper();
- resumable.setLastOffset(clientWrapper.getLatestUpdateSequence());
+ resumable.updateLastOffset(clientWrapper.getLatestUpdateSequence());
}
}
diff --git a/components/camel-file/src/main/java/org/apache/camel/component/file/FileConsumer.java b/components/camel-file/src/main/java/org/apache/camel/component/file/FileConsumer.java
index c4d2d20..e9cca6b 100644
--- a/components/camel-file/src/main/java/org/apache/camel/component/file/FileConsumer.java
+++ b/components/camel-file/src/main/java/org/apache/camel/component/file/FileConsumer.java
@@ -33,6 +33,8 @@ import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.component.file.consumer.FileConsumerResumeStrategy;
import org.apache.camel.component.file.consumer.FileResumeSet;
+import org.apache.camel.component.file.consumer.FileSetResumeStrategy;
+import org.apache.camel.component.file.consumer.GenericFileResumeStrategy;
import org.apache.camel.util.FileUtil;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
@@ -102,9 +104,8 @@ public class FileConsumer extends GenericFileConsumer<File> {
GenericFile<File> gf
= asGenericFile(endpointPath, file, getEndpoint().getCharset(), getEndpoint().isProbeContentType());
- if (resumeStrategy != null) {
- long offset = resumeStrategy.lastOffset(file);
- gf.setLastOffset(offset);
+ if (resumeStrategy != null && resumeStrategy instanceof GenericFileResumeStrategy) {
+ resumeStrategy.resume(gf);
}
if (file.isDirectory()) {
@@ -171,11 +172,11 @@ public class FileConsumer extends GenericFileConsumer<File> {
}
}
- if (resumeStrategy != null) {
+ if (resumeStrategy != null && resumeStrategy instanceof FileSetResumeStrategy) {
FileResumeSet resumeSet = new FileResumeSet(dirFiles);
resumeStrategy.resume(resumeSet);
- return resumeSet.hasResumables() ? resumeSet.resumedFiles() : dirFiles;
+ return resumeSet.hasResumables() ? resumeSet.resumed() : dirFiles;
}
return dirFiles;
diff --git a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFile.java b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFile.java
index e851f9e..11866c3 100644
--- a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFile.java
+++ b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFile.java
@@ -22,8 +22,10 @@ import java.nio.file.Path;
import java.util.Map;
import org.apache.camel.Exchange;
-import org.apache.camel.Resumable;
+import org.apache.camel.Offset;
import org.apache.camel.WrappedFile;
+import org.apache.camel.component.file.consumer.GenericFileResumable;
+import org.apache.camel.resume.Offsets;
import org.apache.camel.util.FileUtil;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.StringHelper;
@@ -33,7 +35,7 @@ import org.slf4j.LoggerFactory;
/**
* Generic File. Specific implementations of a file based endpoint need to provide a File for transfer.
*/
-public class GenericFile<T> implements WrappedFile<T>, Resumable<Long> {
+public class GenericFile<T> implements WrappedFile<T>, GenericFileResumable<T> {
private static final Logger LOG = LoggerFactory.getLogger(GenericFile.class);
private final boolean probeContentType;
@@ -416,13 +418,18 @@ public class GenericFile<T> implements WrappedFile<T>, Resumable<Long> {
}
@Override
- public void setLastOffset(Long offset) {
+ public void updateLastOffset(Long offset) {
this.lastOffset = offset;
}
@Override
- public Long getLastOffset() {
- return lastOffset;
+ public Offset<Long> getLastOffset() {
+ return Offsets.of(lastOffset);
+ }
+
+ @Override
+ public T getAddressable() {
+ return file;
}
/**
diff --git a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConverter.java b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConverter.java
index b854698..63c1edf 100644
--- a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConverter.java
+++ b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConverter.java
@@ -194,7 +194,7 @@ public final class GenericFileConverter {
reader = IOHelper.toReader(f, ExchangeHelper.getCharsetName(exchange));
}
- reader.skip(file.getLastOffset());
+ reader.skip(file.getLastOffset().offset());
return reader;
}
return null;
diff --git a/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileConsumerResumeStrategy.java b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileConsumerResumeStrategy.java
index ccd4c05..fed8000 100644
--- a/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileConsumerResumeStrategy.java
+++ b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileConsumerResumeStrategy.java
@@ -17,28 +17,17 @@
package org.apache.camel.component.file.consumer;
-import java.io.File;
-
import org.apache.camel.ResumeStrategy;
/**
* Defines resume strategy for consumers of the file component.
*/
-public interface FileConsumerResumeStrategy extends ResumeStrategy<FileResumeSet> {
+public interface FileConsumerResumeStrategy<T> extends ResumeStrategy {
/**
* Returns the last offset read for the given file.
- *
- * @param file the file to check for the last offset
- * @return The last offset read or 0 (zero) if none has been read.
- */
- long lastOffset(File file);
-
- /**
- * Perform the resume operation. This runs in the scope of the FileConsumer instance thread.
*
- * @param resumeInfo resume information
+ * @param resumable the resumable file or resumable set to run the resume
*/
- @Override
- void resume(FileResumeSet resumeInfo);
+ void resume(T resumable);
}
diff --git a/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileResumeSet.java b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileResumeSet.java
index e31a8e4..780f77b 100644
--- a/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileResumeSet.java
+++ b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileResumeSet.java
@@ -18,20 +18,21 @@
package org.apache.camel.component.file.consumer;
import java.io.File;
-import java.util.Arrays;
import java.util.Objects;
import java.util.function.Predicate;
+import org.apache.camel.ResumableSet;
+
/**
* This contains the input/output file set for resume operations.
*/
-public final class FileResumeSet {
+public final class FileResumeSet implements ResumableSet<File> {
private final File[] inputFiles;
private File[] outputFiles;
public FileResumeSet(File[] inputFiles) {
- Objects.requireNonNull(inputFiles, "A list of input files must be provided for the resume info");
- this.inputFiles = inputFiles;
+ this.inputFiles = Objects.requireNonNull(inputFiles,
+ "A list of input files must be provided for the resume info");
}
/**
@@ -40,18 +41,7 @@ public final class FileResumeSet {
* @param resumableCheck a checker method that returns true if the file should be resumed or false otherwise
*/
public void resumeEach(Predicate<File> resumableCheck) {
- this.outputFiles = null;
- File[] tmp = Arrays.copyOf(inputFiles, inputFiles.length);
- int count = 0;
-
- for (File file : inputFiles) {
- if (resumableCheck.test(file)) {
- tmp[count] = file;
- count++;
- }
- }
-
- this.outputFiles = Arrays.copyOf(tmp, count);
+ this.outputFiles = resumeEach(inputFiles, resumableCheck);
}
/**
@@ -59,7 +49,7 @@ public final class FileResumeSet {
*
* @return an array with the files that should be resumed
*/
- public File[] resumedFiles() {
+ public File[] resumed() {
return outputFiles;
}
@@ -69,6 +59,6 @@ public final class FileResumeSet {
* @return true if there are resumable files or false otherwise
*/
public boolean hasResumables() {
- return outputFiles != null && outputFiles.length > 0;
+ return outputFiles != null && outputFiles.length != inputFiles.length;
}
}
diff --git a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileSetResumeStrategy.java
similarity index 76%
copy from components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java
copy to components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileSetResumeStrategy.java
index cd8ba08..dc75e80 100644
--- a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java
+++ b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileSetResumeStrategy.java
@@ -15,13 +15,8 @@
* limitations under the License.
*/
-package org.apache.camel.component.couchdb.consumer;
+package org.apache.camel.component.file.consumer;
-import org.apache.camel.ResumeStrategy;
-
-/**
- * Defines a resumable strategy usable by the CouchDB component
- */
-public interface CouchDbResumeStrategy extends ResumeStrategy<CouchDbResumable> {
+public interface FileSetResumeStrategy extends FileConsumerResumeStrategy<FileResumeSet> {
}
diff --git a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/GenericFileResumable.java
similarity index 76%
copy from components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java
copy to components/camel-file/src/main/java/org/apache/camel/component/file/consumer/GenericFileResumable.java
index cd8ba08..037621d 100644
--- a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java
+++ b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/GenericFileResumable.java
@@ -15,13 +15,10 @@
* limitations under the License.
*/
-package org.apache.camel.component.couchdb.consumer;
+package org.apache.camel.component.file.consumer;
-import org.apache.camel.ResumeStrategy;
+import org.apache.camel.Resumable;
-/**
- * Defines a resumable strategy usable by the CouchDB component
- */
-public interface CouchDbResumeStrategy extends ResumeStrategy<CouchDbResumable> {
+public interface GenericFileResumable<T> extends Resumable<T, Long> {
}
diff --git a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/GenericFileResumeStrategy.java
similarity index 76%
copy from components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java
copy to components/camel-file/src/main/java/org/apache/camel/component/file/consumer/GenericFileResumeStrategy.java
index cd8ba08..294e772 100644
--- a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java
+++ b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/GenericFileResumeStrategy.java
@@ -15,13 +15,8 @@
* limitations under the License.
*/
-package org.apache.camel.component.couchdb.consumer;
+package org.apache.camel.component.file.consumer;
-import org.apache.camel.ResumeStrategy;
-
-/**
- * Defines a resumable strategy usable by the CouchDB component
- */
-public interface CouchDbResumeStrategy extends ResumeStrategy<CouchDbResumable> {
+public interface GenericFileResumeStrategy<T> extends FileConsumerResumeStrategy<GenericFileResumable<T>> {
}
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaConsumerResumeStrategy.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaConsumerResumeStrategy.java
index b104bbf..c433745 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaConsumerResumeStrategy.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaConsumerResumeStrategy.java
@@ -18,20 +18,19 @@ package org.apache.camel.component.kafka.consumer.support;
import org.apache.camel.ResumeStrategy;
import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
/**
* Defines a strategy for handling resume operations. Implementations can define different ways to handle how to resume
* processing records.
+ *
+ * The resume runs in the scope of the Kafka Consumer thread and may run concurrently with other consumer instances when
+ * the component is set up to use more than one of them. As such, implementations are responsible for ensuring the
+ * thread-safety of the operations within the resume method.
*/
-public interface KafkaConsumerResumeStrategy extends ResumeStrategy<Consumer<?, ?>> {
+public interface KafkaConsumerResumeStrategy extends ResumeStrategy<KafkaResumable> {
+ void setConsumer(Consumer<?, ?> consumer);
+
+ default void resume(KafkaResumable resumable) {
- /**
- * Perform the resume operation. This runs in the scope of the Kafka Consumer thread and may run concurrently with
- * other consumer instances when the component is set up to use more than one of them. As such, implementations are
- * responsible for ensuring the thread-safety of the operations within the resume method.
- *
- * @param consumer an instance of the KafkaConsumer which is resuming the operation
- */
- void resume(Consumer<?, ?> consumer);
+ }
}
diff --git a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/LatestUpdateSequenceResumeStrategy.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaOffset.java
similarity index 59%
copy from components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/LatestUpdateSequenceResumeStrategy.java
copy to components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaOffset.java
index ac80db8..8a86b65 100644
--- a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/LatestUpdateSequenceResumeStrategy.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaOffset.java
@@ -15,19 +15,25 @@
* limitations under the License.
*/
-package org.apache.camel.component.couchdb.consumer;
+package org.apache.camel.component.kafka.consumer.support;
-import org.apache.camel.component.couchdb.CouchDbClientWrapper;
+import org.apache.camel.Offset;
+import org.apache.camel.util.KeyValueHolder;
/**
- * A resume strategy that resumes from the last update sequence
+ * Offset class for Kafka
*/
-public final class LatestUpdateSequenceResumeStrategy implements CouchDbResumeStrategy {
+public class KafkaOffset implements Offset<KeyValueHolder<String, String>> {
+ private final String topicPartition;
+ private final String offset;
- @Override
- public void resume(CouchDbResumable resumable) {
- CouchDbClientWrapper clientWrapper = resumable.getClientWrapper();
+ public KafkaOffset(String topicPartition, String offset) {
+ this.topicPartition = topicPartition;
+ this.offset = offset;
+ }
- resumable.setLastOffset(clientWrapper.getLatestUpdateSequence());
+ @Override
+ public KeyValueHolder<String, String> offset() {
+ return new KeyValueHolder<>(topicPartition, offset);
}
}
diff --git a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumable.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaResumable.java
similarity index 56%
copy from components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumable.java
copy to components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaResumable.java
index ac3d679..a210c47 100644
--- a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumable.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaResumable.java
@@ -15,39 +15,33 @@
* limitations under the License.
*/
-package org.apache.camel.component.couchdb.consumer;
+package org.apache.camel.component.kafka.consumer.support;
+import org.apache.camel.Offset;
import org.apache.camel.Resumable;
-import org.apache.camel.component.couchdb.CouchDbClientWrapper;
+import org.apache.camel.resume.Offsets;
-/**
- * Wraps the resume data for CouchDb
- */
-public class CouchDbResumable implements Resumable<String> {
- private final CouchDbClientWrapper clientWrapper;
+public class KafkaResumable implements Resumable<String, String> {
+ private String partition;
private String offset;
- public CouchDbResumable(CouchDbClientWrapper clientWrapper, String offset) {
- this.clientWrapper = clientWrapper;
+ public KafkaResumable(String partition, String offset) {
+ this.partition = partition;
this.offset = offset;
}
@Override
- public void setLastOffset(String offset) {
+ public void updateLastOffset(String offset) {
this.offset = offset;
}
@Override
- public String getLastOffset() {
- return offset;
+ public Offset<String> getLastOffset() {
+ return Offsets.of(offset);
}
- /**
- * Gets the client wrapper. Fine for local access, but should be restricted for global access on the API
- *
- * @return the client wrapper
- */
- CouchDbClientWrapper getClientWrapper() {
- return clientWrapper;
+ @Override
+ public String getAddressable() {
+ return partition;
}
}
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/OffsetKafkaConsumerResumeStrategy.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/OffsetKafkaConsumerResumeStrategy.java
index cfe855c..4502831 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/OffsetKafkaConsumerResumeStrategy.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/OffsetKafkaConsumerResumeStrategy.java
@@ -35,11 +35,17 @@ public class OffsetKafkaConsumerResumeStrategy implements KafkaConsumerResumeStr
private static final Logger LOG = LoggerFactory.getLogger(OffsetKafkaConsumerResumeStrategy.class);
private final StateRepository<String, String> offsetRepository;
+ private Consumer<?, ?> consumer;
public OffsetKafkaConsumerResumeStrategy(StateRepository<String, String> offsetRepository) {
this.offsetRepository = offsetRepository;
}
+ @Override
+ public void setConsumer(Consumer<?, ?> consumer) {
+ this.consumer = consumer;
+ }
+
private void resumeFromOffset(final Consumer<?, ?> consumer, TopicPartition topicPartition, String offsetState) {
// The state contains the last read offset, so you need to seek from the next one
long offset = deserializeOffsetValue(offsetState) + 1;
@@ -48,7 +54,7 @@ public class OffsetKafkaConsumerResumeStrategy implements KafkaConsumerResumeStr
}
@Override
- public void resume(final Consumer<?, ?> consumer) {
+ public void resume() {
Set<TopicPartition> assignments = consumer.assignment();
for (TopicPartition topicPartition : assignments) {
String offsetState = offsetRepository.getState(serializeOffsetKey(topicPartition));
@@ -57,4 +63,14 @@ public class OffsetKafkaConsumerResumeStrategy implements KafkaConsumerResumeStr
}
}
}
+
+ /*
+ Note: when self-managing the offsets, we don't need to use the information on the resumable
+ instance. We can collect the assignments directly from the consumer instance as we always did.
+ */
+ @SuppressWarnings("unused")
+ @Override
+ public void resume(KafkaResumable resumable) {
+ resume();
+ }
}
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/PartitionAssignmentListener.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/PartitionAssignmentListener.java
index b034293..efedbe5 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/PartitionAssignmentListener.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/PartitionAssignmentListener.java
@@ -18,8 +18,10 @@
package org.apache.camel.component.kafka.consumer.support;
import java.util.Collection;
+import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
+import java.util.stream.Collectors;
import org.apache.camel.component.kafka.KafkaConfiguration;
import org.apache.camel.component.kafka.consumer.CommitManager;
@@ -53,6 +55,7 @@ public class PartitionAssignmentListener implements ConsumerRebalanceListener {
this.stopStateSupplier = stopStateSupplier;
this.resumeStrategy = ResumeStrategyFactory.newResumeStrategy(configuration);
+ resumeStrategy.setConsumer(consumer);
}
@Override
@@ -91,10 +94,14 @@ public class PartitionAssignmentListener implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+
if (LOG.isDebugEnabled()) {
partitions.forEach(p -> LOG.debug("onPartitionsAssigned: {} from {}", threadId, p.topic()));
+
}
+ List<KafkaResumable> resumables = partitions.stream()
+ .map(p -> new KafkaResumable(String.valueOf(p.partition()), p.topic())).collect(Collectors.toList());
- resumeStrategy.resume(consumer);
+ resumables.forEach(r -> resumeStrategy.resume(r));
}
}
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/ResumeStrategyFactory.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/ResumeStrategyFactory.java
index bd94ccc..3465ef5 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/ResumeStrategyFactory.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/ResumeStrategyFactory.java
@@ -28,10 +28,17 @@ public final class ResumeStrategyFactory {
* A NO-OP resume strategy that does nothing (i.e.: no resume)
*/
private static class NoOpKafkaConsumerResumeStrategy implements KafkaConsumerResumeStrategy {
+
@SuppressWarnings("unused")
@Override
- public void resume(Consumer<?, ?> consumer) {
- // NO-OP
+ public void setConsumer(Consumer<?, ?> consumer) {
+
+ }
+
+ @SuppressWarnings("unused")
+ @Override
+ public void resume() {
+
}
}
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/SeekPolicyKafkaConsumerResumeStrategy.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/SeekPolicyKafkaConsumerResumeStrategy.java
index 573d0b1..91db91b 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/SeekPolicyKafkaConsumerResumeStrategy.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/SeekPolicyKafkaConsumerResumeStrategy.java
@@ -28,13 +28,19 @@ public class SeekPolicyKafkaConsumerResumeStrategy implements KafkaConsumerResum
private static final Logger LOG = LoggerFactory.getLogger(SeekPolicyKafkaConsumerResumeStrategy.class);
private final String seekPolicy;
+ private Consumer<?, ?> consumer;
public SeekPolicyKafkaConsumerResumeStrategy(String seekPolicy) {
this.seekPolicy = seekPolicy;
}
@Override
- public void resume(final Consumer<?, ?> consumer) {
+ public void setConsumer(Consumer<?, ?> consumer) {
+ this.consumer = consumer;
+ }
+
+ @Override
+ public void resume() {
if (seekPolicy.equals("beginning")) {
LOG.debug("Seeking from the beginning of topic");
consumer.seekToBeginning(consumer.assignment());
@@ -43,4 +49,13 @@ public class SeekPolicyKafkaConsumerResumeStrategy implements KafkaConsumerResum
consumer.seekToEnd(consumer.assignment());
}
}
+
+ /*
+ * Note: when using the seek policy strategy, we don't use the resumable information
+ * because we use the consumer to set the policy.
+ */
+ @Override
+ public void resume(KafkaResumable resumable) {
+ resume();
+ }
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerWithResumeStrategyIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerWithResumeStrategyIT.java
index 15bd79c..c9f9b95 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerWithResumeStrategyIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerWithResumeStrategyIT.java
@@ -24,6 +24,7 @@ import org.apache.camel.BindToRegistry;
import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.consumer.support.KafkaConsumerResumeStrategy;
+import org.apache.camel.component.kafka.consumer.support.KafkaResumable;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.kafka.clients.consumer.Consumer;
import org.junit.jupiter.api.AfterEach;
@@ -53,16 +54,23 @@ public class KafkaConsumerWithResumeStrategyIT extends BaseEmbeddedKafkaTestSupp
}
@Override
- public void resume(Consumer<?, ?> consumer) {
- resumeCalled = true;
-
+ public void setConsumer(Consumer<?, ?> consumer) {
if (consumer != null) {
consumerIsNull = false;
}
+ }
+ @Override
+ public void resume(KafkaResumable resumable) {
+ resumeCalled = true;
messagesLatch.countDown();
}
+ @Override
+ public void resume() {
+
+ }
+
public boolean isResumeCalled() {
return resumeCalled;
}
diff --git a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java b/core/camel-api/src/main/java/org/apache/camel/Offset.java
similarity index 77%
copy from components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java
copy to core/camel-api/src/main/java/org/apache/camel/Offset.java
index cd8ba08..1e73d1d 100644
--- a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/consumer/CouchDbResumeStrategy.java
+++ b/core/camel-api/src/main/java/org/apache/camel/Offset.java
@@ -15,13 +15,20 @@
* limitations under the License.
*/
-package org.apache.camel.component.couchdb.consumer;
-
-import org.apache.camel.ResumeStrategy;
+package org.apache.camel;
/**
- * Defines a resumable strategy usable by the CouchDB component
+ * Generic offset without a concrete type
+ *
+ * @param <T> the type of the offset
*/
-public interface CouchDbResumeStrategy extends ResumeStrategy<CouchDbResumable> {
+public interface Offset<T> {
+
+ /**
+ * Gets the offset value
+ *
+ * @return the offset value
+ */
+ T offset();
}
diff --git a/core/camel-api/src/main/java/org/apache/camel/Resumable.java b/core/camel-api/src/main/java/org/apache/camel/Resumable.java
index 35977b3..98c5c28 100644
--- a/core/camel-api/src/main/java/org/apache/camel/Resumable.java
+++ b/core/camel-api/src/main/java/org/apache/camel/Resumable.java
@@ -22,22 +22,30 @@ package org.apache.camel;
* For example, when reading large files, it may be possible to inform the last offset that was read, thus allowing
* users of this interface to skip to that offset. This can potentially improve resumable operations by allowing
* reprocessing of data.
- *
+ *
+ * @param <Y> the type of the key, name or object that can be addressed by the given offset
* @param <T> the type of the addressable value for the resumable object (for example, a file would use a Long value)
*/
-public interface Resumable<T> {
+public interface Resumable<Y, T> {
/**
- * Sets the last offset the last offset as appropriate for the user of the interface
- *
+ * Updates the last offset the last offset as appropriate for the user of the interface
+ *
* @param offset the offset value
*/
- void setLastOffset(T offset);
+ void updateLastOffset(T offset);
/**
* Gets the last offset value
*
* @return the last offset value according to the interface and type implemented
*/
- T getLastOffset();
+ Offset<T> getLastOffset();
+
+ /**
+ * Gets the addressable part (key) of the resumable
+ *
+ * @return the addressable part of the resumable
+ */
+ Y getAddressable();
}
diff --git a/core/camel-api/src/main/java/org/apache/camel/ResumableSet.java b/core/camel-api/src/main/java/org/apache/camel/ResumableSet.java
new file mode 100644
index 0000000..4e4693d
--- /dev/null
+++ b/core/camel-api/src/main/java/org/apache/camel/ResumableSet.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel;
+
+import java.util.Arrays;
+import java.util.function.Predicate;
+
+public interface ResumableSet<T> {
+
+ /**
+ * Iterates over the set of input checking if they should be resumed or not.
+ *
+ * @param input
+ * @param resumableCheck
+ * @return
+ */
+ default T[] resumeEach(T[] input, Predicate<T> resumableCheck) {
+
+ T[] tmp = Arrays.copyOf(input, input.length);
+ int count = 0;
+
+ for (T entry : input) {
+ if (resumableCheck.test(entry)) {
+ tmp[count] = entry;
+ count++;
+ }
+ }
+
+ if (count != input.length) {
+ return Arrays.copyOf(tmp, count);
+ }
+
+ return input;
+ }
+
+ /**
+ * Iterates over the set of input checking if they should be resumed or not
+ *
+ * @param resumableCheck a checker method that returns true if a single entry of the input should be resumed or
+ * false otherwise
+ */
+ void resumeEach(Predicate<T> resumableCheck);
+
+ /**
+ * Gets the input that should be resumed
+ *
+ * @return an array with the input that should be resumed
+ */
+ T[] resumed();
+
+ /**
+ * Whether there are resumable entries to process
+ *
+ * @return true if there are resumable entries or false otherwise
+ */
+ boolean hasResumables();
+}
diff --git a/core/camel-api/src/main/java/org/apache/camel/ResumeStrategy.java b/core/camel-api/src/main/java/org/apache/camel/ResumeStrategy.java
index 17f9828..e8c7f28 100644
--- a/core/camel-api/src/main/java/org/apache/camel/ResumeStrategy.java
+++ b/core/camel-api/src/main/java/org/apache/camel/ResumeStrategy.java
@@ -27,7 +27,6 @@ public interface ResumeStrategy<T> {
* A consumer, iterator or value class that can be used to set the index position from which to resume from. The
* type is specific to the component.
*
- * @param resumable A resumable object.
*/
- void resume(T resumable);
+ void resume();
}
diff --git a/core/camel-api/src/main/java/org/apache/camel/Resumable.java b/core/camel-api/src/main/java/org/apache/camel/UpdatableConsumerResumeStrategy.java
similarity index 57%
copy from core/camel-api/src/main/java/org/apache/camel/Resumable.java
copy to core/camel-api/src/main/java/org/apache/camel/UpdatableConsumerResumeStrategy.java
index 35977b3..34bd716 100644
--- a/core/camel-api/src/main/java/org/apache/camel/Resumable.java
+++ b/core/camel-api/src/main/java/org/apache/camel/UpdatableConsumerResumeStrategy.java
@@ -18,26 +18,19 @@
package org.apache.camel;
/**
- * This provides an interface for resumable objects. Such objects allow its users to address them at a specific offset.
- * For example, when reading large files, it may be possible to inform the last offset that was read, thus allowing
- * users of this interface to skip to that offset. This can potentially improve resumable operations by allowing
- * reprocessing of data.
- *
+ * An updatable resume strategy
+ *
+ * @param <K> the type of the key, name or object that can be addressed by the given offset
* @param <T> the type of the addressable value for the resumable object (for example, a file would use a Long value)
+ * @param <T> a resumable type capable of handling/storing/holding resumable information for the addressable and offset
*/
-public interface Resumable<T> {
-
- /**
- * Sets the last offset the last offset as appropriate for the user of the interface
- *
- * @param offset the offset value
- */
- void setLastOffset(T offset);
+public interface UpdatableConsumerResumeStrategy<K, V, T extends Resumable<K, V>> {
/**
- * Gets the last offset value
+ * Updates the last processed offset
*
- * @return the last offset value according to the interface and type implemented
+ * @param offset the offset to update
+ * @throws Exception if unable to update the offset
*/
- T getLastOffset();
+ void updateLastOffset(T offset) throws Exception;
}
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerResumeFromOffsetStrategyTest.java b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerResumeFromOffsetStrategyTest.java
index ac18a95..62c4560 100644
--- a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerResumeFromOffsetStrategyTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerResumeFromOffsetStrategyTest.java
@@ -22,30 +22,26 @@ import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.file.consumer.FileConsumerResumeStrategy;
-import org.apache.camel.component.file.consumer.FileResumeSet;
+import org.apache.camel.component.file.consumer.GenericFileResumable;
+import org.apache.camel.component.file.consumer.GenericFileResumeStrategy;
import org.apache.camel.component.mock.MockEndpoint;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class FileConsumerResumeFromOffsetStrategyTest extends ContextTestSupport {
- private static class TestResumeStrategy implements FileConsumerResumeStrategy {
- private static final Logger LOG = LoggerFactory.getLogger(TestResumeStrategy.class);
-
+ private static class TestResumeStrategy implements GenericFileResumeStrategy<File> {
@Override
- public long lastOffset(File file) {
- if (!file.getName().startsWith("resume-from-offset")) {
+ public void resume(GenericFileResumable<File> resumable) {
+ if (!resumable.getAddressable().getName().startsWith("resume-from-offset")) {
throw new RuntimeCamelException("Invalid file - resume strategy should not have been called!");
}
- return 3;
+ resumable.updateLastOffset(3L);
}
@Override
- public void resume(FileResumeSet resumeSet) {
+ public void resume() {
// NO-OP
}
}
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerResumeStrategyTest.java b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerResumeStrategyTest.java
index 0269485..129399c 100644
--- a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerResumeStrategyTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerResumeStrategyTest.java
@@ -16,36 +16,36 @@
*/
package org.apache.camel.component.file;
-import java.io.File;
import java.util.Arrays;
import java.util.List;
+import java.util.Objects;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.file.consumer.FileConsumerResumeStrategy;
import org.apache.camel.component.file.consumer.FileResumeSet;
+import org.apache.camel.component.file.consumer.FileSetResumeStrategy;
import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.util.IOHelper;
import org.junit.jupiter.api.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class FileConsumerResumeStrategyTest extends ContextTestSupport {
- private static class TestResumeStrategy implements FileConsumerResumeStrategy {
- private static final Logger LOG = LoggerFactory.getLogger(TestResumeStrategy.class);
+ private static class TestResumeStrategy implements FileSetResumeStrategy {
+ private List<String> processedFiles = Arrays.asList("0.txt", "1.txt", "2.txt");
+ private FileResumeSet resumeSet;
@Override
- public long lastOffset(File file) {
- return IOHelper.INITIAL_OFFSET;
+ public void resume(FileResumeSet resumeSet) {
+ this.resumeSet = Objects.requireNonNull(resumeSet);
+
+ resume();
}
@Override
- public void resume(FileResumeSet resumeSet) {
- List<String> processedFiles = Arrays.asList("0.txt", "1.txt", "2.txt");
-
- resumeSet.resumeEach(f -> !processedFiles.contains(f.getName()));
+ public void resume() {
+ if (resumeSet != null) {
+ resumeSet.resumeEach(f -> !processedFiles.contains(f.getName()));
+ }
}
}
@@ -75,7 +75,8 @@ public class FileConsumerResumeStrategyTest extends ContextTestSupport {
bindToRegistry("testResumeStrategy", new TestResumeStrategy());
from(fileUri("resume?noop=true&recursive=true&resumeStrategy=#testResumeStrategy"))
- .convertBodyTo(String.class).to("mock:result");
+ .convertBodyTo(String.class)
+ .to("mock:result");
}
};
}
diff --git a/core/camel-support/src/main/java/org/apache/camel/resume/Offsets.java b/core/camel-support/src/main/java/org/apache/camel/resume/Offsets.java
new file mode 100644
index 0000000..a7f4683
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/resume/Offsets.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.resume;
+
+import org.apache.camel.Offset;
+
+/**
+ * Offset handling support
+ */
+public final class Offsets {
+ /**
+ * Default initial offset when using long offsets
+ */
+ public static final Offset<Long> INITIAL_LONG = Offsets.of(0L);
+
+ private Offsets() {
+
+ }
+
+ /**
+ * Creates a new offset with the given offset value
+ *
+ * @param offsetValue the offset value
+ * @param <T> The type of the offset
+ * @return A new Offset holder with the given offset value
+ */
+ public static <T> Offset<T> of(T offsetValue) {
+ return () -> offsetValue;
+ }
+
+ /**
+ * Creates a new offset with a default value in case of the offset value is null
+ *
+ * @param offsetValue the offset value
+ * @param defaultValue the default offset value to use if the provided offset value is null
+ * @param <T> the type of the offset
+ * @return A new Offset holder with the given offset value or the default one if the offset value is
+ * null
+ */
+ public static <T> Offset<T> ofNullable(T offsetValue, T defaultValue) {
+ if (offsetValue != null) {
+ return () -> offsetValue;
+ }
+
+ return Offsets.of(defaultValue);
+ }
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/resume/Resumables.java b/core/camel-support/src/main/java/org/apache/camel/resume/Resumables.java
new file mode 100644
index 0000000..38bfed3
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/resume/Resumables.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.resume;
+
+import org.apache.camel.Offset;
+import org.apache.camel.Resumable;
+
+/**
+ * A wrapper for resumable entities
+ */
+public final class Resumables {
+
+ /**
+ * An anonymous resumable type
+ *
+ * @param <K> the type of the key, name or object that can be addressed by the given offset (aka addressable)
+ * @param <V> the type of offset
+ */
+ private static class AnonymousResumable<K, V> implements Resumable<K, V> {
+ private final K addressable;
+ private V offset;
+
+ /**
+ * Creates a new anonymous resumable type
+ *
+ * @param addressable the key, name or object that can be addressed by the given offset
+ */
+ public AnonymousResumable(K addressable) {
+ this.addressable = addressable;
+ }
+
+ /**
+ * Creates a new anonymous resumable type
+ *
+ * @param addressable the key, name or object that can be addressed by the given offset
+ * @param offset the offset value
+ */
+ public AnonymousResumable(K addressable, V offset) {
+ this.addressable = addressable;
+ this.offset = offset;
+ }
+
+ @Override
+ public void updateLastOffset(V offset) {
+ this.offset = offset;
+ }
+
+ @Override
+ public Offset<V> getLastOffset() {
+ return Offsets.of(offset);
+ }
+
+ @Override
+ public K getAddressable() {
+ return addressable;
+ }
+ }
+
+ private Resumables() {
+
+ }
+
+ /**
+ * Creates a new resumable for an addressable
+ *
+ * @param addressable the key, name or object that can be addressed by the given offset
+ * @param offset the offset value
+ * @param <K> the type of the key, name or object that can be addressed by the given offset (aka
+ * addressable)
+ * @param <V> the type of offset
+ * @return A new resumable entity for the given addressable with the given offset value
+ */
+ public static <K, V> Resumable<K, V> of(K addressable, V offset) {
+ return new AnonymousResumable<>(addressable, offset);
+ }
+}