Adjust the CamelSourceTask code to reduce CPU usage and heap allocations
This modifies the code so that it blocks while waiting for the messages
to arrive while also respecting the maxPollDuration.
@@ -119,17 +119,28 @@ public class CamelSourceTask extends SourceTask {
}
}
+ private long remaining(long startPollEpochMilli, long maxPollDuration) {
+ return maxPollDuration - (Instant.now().toEpochMilli() - startPollEpochMilli);
+ }
+
+
@Override
public synchronized List<SourceRecord> poll() {
- long startPollEpochMilli = Instant.now().toEpochMilli();
+ final long startPollEpochMilli = Instant.now().toEpochMilli();
+
+ long remaining = remaining(startPollEpochMilli, maxPollDuration);
long collectedRecords = 0L;
List<SourceRecord> records = new ArrayList<>();
- while (collectedRecords < maxBatchPollSize && (Instant.now().toEpochMilli() - startPollEpochMilli) < maxPollDuration) {
- Exchange exchange = consumer.receiveNoWait();
+ while (collectedRecords < maxBatchPollSize && remaining > 0) {
+ Exchange exchange = consumer.receive(remaining);
+ if (exchange == null) {
+ // Nothing received, abort and return what we received so far
+ break;
+ }
- if (exchange != null) {
- LOG.debug("Received Exchange {} with Message {} from Endpoint {}", exchange.getExchangeId(), exchange.getMessage().getMessageId(), exchange.getFromEndpoint());
+ LOG.debug("Received Exchange {} with Message {} from Endpoint {}", exchange.getExchangeId(),
+ exchange.getMessage().getMessageId(), exchange.getFromEndpoint());
// TODO: see if there is a better way to use sourcePartition
// an sourceOffset
@@ -143,7 +154,9 @@ public class CamelSourceTask extends SourceTask {
final Schema messageBodySchema = messageBodyValue != null ? SchemaHelper.buildSchemaBuilderForType(messageBodyValue) : null;
for (String singleTopic : topics) {
- SourceRecord record = new SourceRecord(sourcePartition, sourceOffset, singleTopic, messageKeySchema, messageHeaderKey, messageBodySchema, messageBodyValue);
+ SourceRecord record = new SourceRecord(sourcePartition, sourceOffset, singleTopic, messageKeySchema,
+ messageHeaderKey, messageBodySchema, messageBodyValue);
+
if (exchange.getMessage().hasHeaders()) {
setAdditionalHeaders(record, exchange.getMessage().getHeaders(), HEADER_CAMEL_PREFIX);
}
@@ -155,17 +168,10 @@ public class CamelSourceTask extends SourceTask {
records.add(record);
}
collectedRecords++;
- } else {
- break;
- }
- }
-
- if (records.isEmpty()) {
- return Collections.EMPTY_LIST;
- } else {
- return records;
+ remaining = remaining(startPollEpochMilli, maxPollDuration);
}
+ return records.isEmpty() ? null : records;
}
@Override
