You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "CodeSmell (via GitHub)" <gi...@apache.org> on 2023/11/07 23:10:41 UTC
[PR] CAMEL-20044 [camel]
CodeSmell opened a new pull request, #11935:
URL: https://github.com/apache/camel/pull/11935
# Description
This PR offers a fix for breakOnFirstError where the incorrect partition:offset when a forceCommit is done
# Target
This PR is off the camel-3.21.x branch
# Tracking
CAMEL-20044
CAMEL-19894
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "CodeSmell (via GitHub)" <gi...@apache.org>.
CodeSmell commented on code in PR #11935:
URL: https://github.com/apache/camel/pull/11935#discussion_r1388283099
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaRecordProcessor.java:
##########
@@ -149,14 +146,30 @@ private boolean processException(
Exception exc = exchange.getException();
LOG.warn("Error during processing {} from topic: {} due to {}", exchange, topicPartition.topic(),
exc.getMessage());
- LOG.warn("Will seek consumer to offset {} on partition {} and start polling again.",
- lastResult.getPartitionLastOffset(), lastResult.getPartition());
+ LOG.warn("Will seek consumer to offset {} on partition {} and start polling again.",
+ record.offset(), record.partition());
}
// force commit, so we resume on next poll where we failed
// except when the failure happened at the first message in a poll
if (lastResult.getPartitionLastOffset() != AbstractCommitManager.START_OFFSET) {
- commitManager.forceCommit(topicPartition, lastResult.getPartitionLastOffset());
+ // the record we are processing had the error
+ // so we will force commit the offset prior
+ // this will enable the current desired behavior to
+ // retry the message 1 more time
+ //
+ // Note: without a more extensive look at handling of breakOnFirstError
+ // we will still need the lastResult so that we don't force
+ // retrying this message over and over
+ // commitManager.forceCommit(topicPartition, record.offset() - 1);
+
+ // we should just do a commit (vs the original forceCommit)
+ // when route uses NOOP Commit Manager it will rely
+ // on the route implementation to explicitly commit offset
+ // when route uses Synch/Asynch Commit Manager it will
+ // ALWAYS commit the offset for the failing record
+ // and will ALWAYS retry it
+ commitManager.commit(topicPartition);
Review Comment:
This is the key change in behavior
It goes beyond a fix and gives route implementers ability to choose how breakOnFirstError should work using the commit manager
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11935:
URL: https://github.com/apache/camel/pull/11935#issuecomment-1802205179
:no_entry_sign: There are (likely) no changes in core core to be tested in this PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "orpiske (via GitHub)" <gi...@apache.org>.
orpiske commented on code in PR #11935:
URL: https://github.com/apache/camel/pull/11935#discussion_r1386300217
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaRecordProcessorFacade.java:
##########
@@ -77,6 +77,9 @@ public ProcessingResult processPolledRecords(
lastResult = processRecord(partition, partitionIterator.hasNext(), recordIterator.hasNext(), lastResult,
kafkaRecordProcessor, record);
+
+ LOG.debug("processed record on partition {} and offset {} and got result for partition {} and offset {}",
Review Comment:
Same note.
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaRecordProcessor.java:
##########
@@ -112,30 +115,56 @@ public ProcessingResult processExchange(
exchange.setException(e);
}
if (exchange.getException() != null) {
- boolean breakOnErrorExit = processException(exchange, partition, lastResult.getPartitionLastOffset(),
+
+ LOG.debug("an exception was thrown for record at partition {} and offset {}",
Review Comment:
Minor thing for consistency: capitalize the first letter of the log message. This is pattern we use throughout the code base.
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaRecordProcessor.java:
##########
@@ -112,30 +115,56 @@ public ProcessingResult processExchange(
exchange.setException(e);
}
if (exchange.getException() != null) {
- boolean breakOnErrorExit = processException(exchange, partition, lastResult.getPartitionLastOffset(),
+
+ LOG.debug("an exception was thrown for record at partition {} and offset {}",
+ record.partition(), record.offset());
+
+ boolean breakOnErrorExit = processException(exchange, topicPartition, record, lastResult,
exceptionHandler);
- return new ProcessingResult(breakOnErrorExit, lastResult.getPartitionLastOffset(), true);
+
+ return new ProcessingResult(breakOnErrorExit, lastResult.getPartition(), lastResult.getPartitionLastOffset(), true);
} else {
- return new ProcessingResult(false, record.offset(), exchange.getException() != null);
+ return new ProcessingResult(false, record.partition(), record.offset(), exchange.getException() != null);
}
}
private boolean processException(
- Exchange exchange, TopicPartition partition, long partitionLastOffset,
+ Exchange exchange, TopicPartition topicPartition,
+ ConsumerRecord<Object, Object> record, ProcessingResult lastResult,
ExceptionHandler exceptionHandler) {
// processing failed due to an unhandled exception, what should we do
if (configuration.isBreakOnFirstError()) {
+
+ if (lastResult.getPartition() != -1 &&
+ lastResult.getPartition() != record.partition()) {
+ LOG.error("about to process an exception with UNEXPECTED partition & offset. Got topic partition {}. " +
Review Comment:
Same note.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "orpiske (via GitHub)" <gi...@apache.org>.
orpiske commented on code in PR #11935:
URL: https://github.com/apache/camel/pull/11935#discussion_r1386662504
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaRecordProcessor.java:
##########
@@ -112,30 +115,56 @@ public ProcessingResult processExchange(
exchange.setException(e);
}
if (exchange.getException() != null) {
- boolean breakOnErrorExit = processException(exchange, partition, lastResult.getPartitionLastOffset(),
+
+ LOG.debug("an exception was thrown for record at partition {} and offset {}",
+ record.partition(), record.offset());
+
+ boolean breakOnErrorExit = processException(exchange, topicPartition, record, lastResult,
exceptionHandler);
- return new ProcessingResult(breakOnErrorExit, lastResult.getPartitionLastOffset(), true);
+
+ return new ProcessingResult(breakOnErrorExit, lastResult.getPartition(), lastResult.getPartitionLastOffset(), true);
} else {
- return new ProcessingResult(false, record.offset(), exchange.getException() != null);
+ return new ProcessingResult(false, record.partition(), record.offset(), exchange.getException() != null);
}
}
private boolean processException(
- Exchange exchange, TopicPartition partition, long partitionLastOffset,
+ Exchange exchange, TopicPartition topicPartition,
+ ConsumerRecord<Object, Object> record, ProcessingResult lastResult,
ExceptionHandler exceptionHandler) {
// processing failed due to an unhandled exception, what should we do
if (configuration.isBreakOnFirstError()) {
+
+ if (lastResult.getPartition() != -1 &&
+ lastResult.getPartition() != record.partition()) {
+ LOG.error("about to process an exception with UNEXPECTED partition & offset. Got topic partition {}. " +
+ " The last result was on partition {} with offset {} but was expecting partition {} with offset {}",
+ topicPartition.partition(), lastResult.getPartition(), lastResult.getPartitionLastOffset(),
+ record.partition(), record.offset());
+ }
+
// we are failing and we should break out
if (LOG.isWarnEnabled()) {
- LOG.warn("Error during processing {} from topic: {}", exchange, partition.topic(), exchange.getException());
- LOG.warn("Will seek consumer to offset {} and start polling again.", partitionLastOffset);
+ LOG.warn("Error during processing {} from topic: {} due to {}", exchange, topicPartition.topic(), exchange.getException());
+ LOG.warn("Will seek consumer to offset {} on partition {} and start polling again.",
+ record.offset(), record.partition());
}
- // force commit, so we resume on next poll where we failed except when the failure happened
- // at the first message in a poll
- if (partitionLastOffset != AbstractCommitManager.START_OFFSET) {
- commitManager.forceCommit(partition, partitionLastOffset);
+ // force commit, so we resume on next poll where we failed
+ // except when the failure happened at the first message in a poll
+ if (lastResult.getPartitionLastOffset() != AbstractCommitManager.START_OFFSET) {
+ // the record we are processing had the error
+ // so we will force commit the offset prior
+ // this will enable the current desired behavior to
+ // retry the message 1 more time
+ //
+ // Note: without a more extensive look at handling of breakOnFirstError
+ // we will still need the lastResult so that we don't force
+ // retrying this message over and over
+ //if (configuration.isBreakOnFirstErrorWithRetry()) {
+ commitManager.forceCommit(topicPartition, record.offset() - 1);
Review Comment:
Understood. I think that's fair. IMHO, it will be important to include that on the documentation as well, so users can refer to this behavior and understand what to expect.
wdyt?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "CodeSmell (via GitHub)" <gi...@apache.org>.
CodeSmell commented on code in PR #11935:
URL: https://github.com/apache/camel/pull/11935#discussion_r1386689966
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaRecordProcessor.java:
##########
@@ -112,30 +115,56 @@ public ProcessingResult processExchange(
exchange.setException(e);
}
if (exchange.getException() != null) {
- boolean breakOnErrorExit = processException(exchange, partition, lastResult.getPartitionLastOffset(),
+
+ LOG.debug("an exception was thrown for record at partition {} and offset {}",
+ record.partition(), record.offset());
+
+ boolean breakOnErrorExit = processException(exchange, topicPartition, record, lastResult,
exceptionHandler);
- return new ProcessingResult(breakOnErrorExit, lastResult.getPartitionLastOffset(), true);
+
+ return new ProcessingResult(breakOnErrorExit, lastResult.getPartition(), lastResult.getPartitionLastOffset(), true);
} else {
- return new ProcessingResult(false, record.offset(), exchange.getException() != null);
+ return new ProcessingResult(false, record.partition(), record.offset(), exchange.getException() != null);
}
}
private boolean processException(
- Exchange exchange, TopicPartition partition, long partitionLastOffset,
+ Exchange exchange, TopicPartition topicPartition,
+ ConsumerRecord<Object, Object> record, ProcessingResult lastResult,
ExceptionHandler exceptionHandler) {
// processing failed due to an unhandled exception, what should we do
if (configuration.isBreakOnFirstError()) {
+
+ if (lastResult.getPartition() != -1 &&
+ lastResult.getPartition() != record.partition()) {
+ LOG.error("about to process an exception with UNEXPECTED partition & offset. Got topic partition {}. " +
+ " The last result was on partition {} with offset {} but was expecting partition {} with offset {}",
+ topicPartition.partition(), lastResult.getPartition(), lastResult.getPartitionLastOffset(),
+ record.partition(), record.offset());
+ }
+
// we are failing and we should break out
if (LOG.isWarnEnabled()) {
- LOG.warn("Error during processing {} from topic: {}", exchange, partition.topic(), exchange.getException());
- LOG.warn("Will seek consumer to offset {} and start polling again.", partitionLastOffset);
+ LOG.warn("Error during processing {} from topic: {} due to {}", exchange, topicPartition.topic(), exchange.getException());
+ LOG.warn("Will seek consumer to offset {} on partition {} and start polling again.",
+ record.offset(), record.partition());
}
- // force commit, so we resume on next poll where we failed except when the failure happened
- // at the first message in a poll
- if (partitionLastOffset != AbstractCommitManager.START_OFFSET) {
- commitManager.forceCommit(partition, partitionLastOffset);
+ // force commit, so we resume on next poll where we failed
+ // except when the failure happened at the first message in a poll
+ if (lastResult.getPartitionLastOffset() != AbstractCommitManager.START_OFFSET) {
+ // the record we are processing had the error
+ // so we will force commit the offset prior
+ // this will enable the current desired behavior to
+ // retry the message 1 more time
+ //
+ // Note: without a more extensive look at handling of breakOnFirstError
+ // we will still need the lastResult so that we don't force
+ // retrying this message over and over
+ //if (configuration.isBreakOnFirstErrorWithRetry()) {
+ commitManager.forceCommit(topicPartition, record.offset() - 1);
Review Comment:
I am good w/ what Camel team thinks is best.
The change here using `forceCommit` will fix the 2 reported issues so would like to see at least that move forward if you don't see any problems.
Looking at the code, if we did make some more changes, we may be able to remove the `lastResult` object that is passed around as well since we have access to the Kafka record.
I can update the docs w/ whatever the new behavior is but may need a pointer as to where/how to do that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11935:
URL: https://github.com/apache/camel/pull/11935#issuecomment-1803566677
### Components test results:
| Total | Tested | Failed :x: | Passed :white_check_mark: |
| --- | --- | --- | --- |
| 1 | 1 | 1 | 0 |
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11935:
URL: https://github.com/apache/camel/pull/11935#issuecomment-1802205153
### Components test results:
| Total | Tested | Failed :x: | Passed :white_check_mark: |
| --- | --- | --- | --- |
| 2 | 2 | 1 | 1 |
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "CodeSmell (via GitHub)" <gi...@apache.org>.
CodeSmell commented on code in PR #11935:
URL: https://github.com/apache/camel/pull/11935#discussion_r1388146069
##########
components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaBreakOnFirstErrorSeekIssueIT.java:
##########
@@ -0,0 +1,146 @@
+package org.apache.camel.component.kafka.integration;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.kafka.MockConsumerInterceptor;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.util.CamelKafkaUtil;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+
+/**
+ * this will test breakOnFirstError functionality
+ * and the issue that was surfaced in CAMEL-19894
+ * regarding failure to correctly commit the offset
+ * in a batch using the Synch Commit Manager
+ *
+ * mimics the reproduction of the problem in
+ * https://github.com/Krivda/camel-bug-reproduction
+ */
+class KafkaBreakOnFirstErrorSeekIssueIT extends BaseEmbeddedKafkaTestSupport {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaBreakOnFirstErrorSeekIssueIT.class);
+
+ public static final String ROUTE_ID = "breakOnFirstError-19894";
+ public static final String TOPIC = "test-foobar";
+
+ @EndpointInject("kafka:" + TOPIC
+ + "?groupId=KafkaBreakOnFirstErrorIT"
+ + "&autoOffsetReset=earliest"
+ + "&autoCommitEnable=false"
+ + "&allowManualCommit=true"
+ + "&breakOnFirstError=true"
+ + "&maxPollRecords=8"
+ + "&pollTimeoutMs=1000"
+ + "&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&kafkaManualCommitFactory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualCommitFactory"
+ + "&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
+ private Endpoint from;
+
+ @EndpointInject("mock:result")
+ private MockEndpoint to;
+
+ private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
+
+ @BeforeEach
+ public void init() {
+ // create the topic w/ 2 partitions
+ final NewTopic mytopic = new NewTopic(TOPIC, 2, (short) 1);
+ kafkaAdminClient.createTopics(Collections.singleton(mytopic));
+
+ // setup the producer
+ Properties props = getDefaultProperties();
+ producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
+ MockConsumerInterceptor.recordsCaptured.clear();
+ }
+
+ @AfterEach
+ public void after() {
+ if (producer != null) {
+ producer.close();
+ }
+ // clean all test topics
+ kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC)).all();
+ }
+
+ @Test
+ void test_camel_19894_test_fix() throws Exception {
+ to.reset();
+ // will consume the payloads from partition 0
+ // and will continually retry the payload with "5"
+ to.expectedMessageCount(4);
+ to.expectedBodiesReceived("1", "2", "3", "4");
+
+ context.getRouteController().stopRoute(ROUTE_ID);
+
+ this.publishMessagesToKafka();
+
+ context.getRouteController().startRoute(ROUTE_ID);
+
+ Awaitility.await()
+ .timeout(30, TimeUnit.SECONDS)
+ .pollDelay(20, TimeUnit.SECONDS)
+ .until(() -> to.getExchanges().size() == 4);
+
+
+ to.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+
+ @Override
+ public void configure() {
+ from(from)
+ .routeId(ROUTE_ID)
+ .autoStartup(false)
+ .process(exchange -> {
+ LOG.debug(CamelKafkaUtil.buildKafkaLogMessage("Consuming", exchange, true));
+ })
+ .process(exchange -> {
+ ifIsFifthRecordThrowException(exchange);
+ })
+ .to(to)
+ .end();
+ }
+ };
+ }
+
+ private void ifIsFifthRecordThrowException(Exchange e) {
+ if (e.getMessage().getBody().equals("5")) {
+ throw new RuntimeException("ERROR_TRIGGERED_BY_TEST");
+ }
+ }
+
+ private void publishMessagesToKafka() {
+ final List<String> producedRecordsPartition0 = List.of("5", "6", "7", "8"); //, "9", "10", "11");
Review Comment:
This test isn't complete as I don't think increasing the partitions is working
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "CodeSmell (via GitHub)" <gi...@apache.org>.
CodeSmell commented on PR #11935:
URL: https://github.com/apache/camel/pull/11935#issuecomment-1804329851
Moved changes to a separate PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11935:
URL: https://github.com/apache/camel/pull/11935#issuecomment-1801556163
:no_entry_sign: There are (likely) no changes in core core to be tested in this PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "CodeSmell (via GitHub)" <gi...@apache.org>.
CodeSmell closed pull request #11935: CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset
URL: https://github.com/apache/camel/pull/11935
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "orpiske (via GitHub)" <gi...@apache.org>.
orpiske commented on PR #11935:
URL: https://github.com/apache/camel/pull/11935#issuecomment-1803485852
@CodeSmell now that we have that other ticket merged, I'll focus on this. I haven't had the time to trigger a test on this on my CI, but I'll be doing this today.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 [camel]
Posted by "CodeSmell (via GitHub)" <gi...@apache.org>.
CodeSmell commented on PR #11935:
URL: https://github.com/apache/camel/pull/11935#issuecomment-1800349018
Note this fix was done off my local branch that has the logging statements.
Those logging statements are also in this approved PR but it wasn't merged yet
- https://github.com/apache/camel/pull/11920
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "CodeSmell (via GitHub)" <gi...@apache.org>.
CodeSmell commented on code in PR #11935:
URL: https://github.com/apache/camel/pull/11935#discussion_r1388285546
##########
components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaBreakOnFirstErrorWithBatchUsingKafkaManualCommitIT.java:
##########
@@ -0,0 +1,159 @@
+package org.apache.camel.component.kafka.integration;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.component.kafka.MockConsumerInterceptor;
+import org.apache.camel.component.kafka.consumer.KafkaManualCommit;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.util.CamelKafkaUtil;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+/**
+ * this will test basic breakOnFirstError functionality
+ * uses allowManualCommit and KafkaManualCommit
+ * because relying on Camel default to use NOOP Commit Manager
+ * this means the route implementation MUST manage all offset commits
+ */
+class KafkaBreakOnFirstErrorWithBatchUsingKafkaManualCommitIT extends BaseEmbeddedKafkaTestSupport {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaBreakOnFirstErrorWithBatchUsingKafkaManualCommitIT.class);
+
+ public static final String ROUTE_ID = "breakOnFirstErrorBatchOnExceptionIT";
+ public static final String TOPIC = "test-foobar";
+
+ @EndpointInject("kafka:" + TOPIC
+ + "?groupId=KafkaBreakOnFirstErrorIT"
+ + "&autoOffsetReset=earliest"
+ + "&autoCommitEnable=false"
+ + "&allowManualCommit=true"
+ + "&breakOnFirstError=true"
+ + "&maxPollRecords=3"
+ + "&pollTimeoutMs=1000"
+ + "&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
+ private Endpoint from;
+
+ @EndpointInject("mock:result")
+ private MockEndpoint to;
+
+ private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
+
+ @BeforeEach
+ public void before() {
+ Properties props = getDefaultProperties();
+ producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
+ MockConsumerInterceptor.recordsCaptured.clear();
+ }
+
+ @AfterEach
+ public void after() {
+ if (producer != null) {
+ producer.close();
+ }
+ // clean all test topics
+ kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC)).all();
+ }
+
+ /**
+ * will continue to retry the message that is in error
+ */
+ @Test
+ public void kafkaBreakOnFirstErrorBasicCapabilityWithoutOnExcepton() throws Exception {
+ to.reset();
+ to.expectedMessageCount(7);
+
+ // old behavior before the fix
+ // message-3 causes an error
+ // and breakOnFirstError will cause it to be retried 1x
+ // then we move on
+ //to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-3", "message-4", "message-5");
Review Comment:
IMO this new behavior is what the route really wants
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11935:
URL: https://github.com/apache/camel/pull/11935#issuecomment-1800348473
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:warning: Please note that the changes on this PR may be **tested automatically**.
If necessary Apache Camel Committers may access logs and test results in the job summaries!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "CodeSmell (via GitHub)" <gi...@apache.org>.
CodeSmell commented on code in PR #11935:
URL: https://github.com/apache/camel/pull/11935#discussion_r1386624448
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaRecordProcessor.java:
##########
@@ -112,30 +115,56 @@ public ProcessingResult processExchange(
exchange.setException(e);
}
if (exchange.getException() != null) {
- boolean breakOnErrorExit = processException(exchange, partition, lastResult.getPartitionLastOffset(),
+
+ LOG.debug("an exception was thrown for record at partition {} and offset {}",
+ record.partition(), record.offset());
+
+ boolean breakOnErrorExit = processException(exchange, topicPartition, record, lastResult,
exceptionHandler);
- return new ProcessingResult(breakOnErrorExit, lastResult.getPartitionLastOffset(), true);
+
+ return new ProcessingResult(breakOnErrorExit, lastResult.getPartition(), lastResult.getPartitionLastOffset(), true);
} else {
- return new ProcessingResult(false, record.offset(), exchange.getException() != null);
+ return new ProcessingResult(false, record.partition(), record.offset(), exchange.getException() != null);
}
}
private boolean processException(
- Exchange exchange, TopicPartition partition, long partitionLastOffset,
+ Exchange exchange, TopicPartition topicPartition,
+ ConsumerRecord<Object, Object> record, ProcessingResult lastResult,
ExceptionHandler exceptionHandler) {
// processing failed due to an unhandled exception, what should we do
if (configuration.isBreakOnFirstError()) {
+
+ if (lastResult.getPartition() != -1 &&
+ lastResult.getPartition() != record.partition()) {
+ LOG.error("about to process an exception with UNEXPECTED partition & offset. Got topic partition {}. " +
+ " The last result was on partition {} with offset {} but was expecting partition {} with offset {}",
+ topicPartition.partition(), lastResult.getPartition(), lastResult.getPartitionLastOffset(),
+ record.partition(), record.offset());
+ }
+
// we are failing and we should break out
if (LOG.isWarnEnabled()) {
- LOG.warn("Error during processing {} from topic: {}", exchange, partition.topic(), exchange.getException());
- LOG.warn("Will seek consumer to offset {} and start polling again.", partitionLastOffset);
+ LOG.warn("Error during processing {} from topic: {} due to {}", exchange, topicPartition.topic(), exchange.getException());
+ LOG.warn("Will seek consumer to offset {} on partition {} and start polling again.",
+ record.offset(), record.partition());
}
- // force commit, so we resume on next poll where we failed except when the failure happened
- // at the first message in a poll
- if (partitionLastOffset != AbstractCommitManager.START_OFFSET) {
- commitManager.forceCommit(partition, partitionLastOffset);
+ // force commit, so we resume on next poll where we failed
+ // except when the failure happened at the first message in a poll
+ if (lastResult.getPartitionLastOffset() != AbstractCommitManager.START_OFFSET) {
+ // the record we are processing had the error
+ // so we will force commit the offset prior
+ // this will enable the current desired behavior to
+ // retry the message 1 more time
+ //
+ // Note: without a more extensive look at handling of breakOnFirstError
+ // we will still need the lastResult so that we don't force
+ // retrying this message over and over
+ //if (configuration.isBreakOnFirstErrorWithRetry()) {
+ commitManager.forceCommit(topicPartition, record.offset() - 1);
Review Comment:
@orpiske
some thoughts here on this
The `forceCommit` was what we had here so kept that for now.
But based on some of the other conversations I think it may be worth changing that to `commit`.
That would defer how `breakOnFirstError` would work based on the `commitManager`
- no op would mean the route implementation would have to handle the offset using `KafkaManualCommit`
- using synch/asynch would mean that the record with the error would be retried
the retry will vary still based on other configuration which is what I think contributes to this flag being so complex and why there are many opinions on how it should work (CAMEL-20089)
This may be a way to alter some of the behavior in 3.x until a refactor in 4.x
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 [camel]
Posted by "CodeSmell (via GitHub)" <gi...@apache.org>.
CodeSmell commented on PR #11935:
URL: https://github.com/apache/camel/pull/11935#issuecomment-1800351119
I was adding integration tests for this fix
The two in the initial commit cover the basic behavior.
Still working on adding IT for the 2 issues
(Note: need to create multiple partitions in the `BaseEmbeddedKafkaTestSupport` but wasn't sure how to do that)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "orpiske (via GitHub)" <gi...@apache.org>.
orpiske commented on PR #11935:
URL: https://github.com/apache/camel/pull/11935#issuecomment-1801456205
Thanks for your contribution!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11935:
URL: https://github.com/apache/camel/pull/11935#issuecomment-1803566714
:no_entry_sign: There are (likely) no changes in core core to be tested in this PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-20044 BreakOnFirstError can commit the wrong partition:offset [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11935:
URL: https://github.com/apache/camel/pull/11935#issuecomment-1801556117
### Components test results:
| Total | Tested | Failed :x: | Passed :white_check_mark: |
| --- | --- | --- | --- |
| 1 | 1 | 1 | 0 |
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org