You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/08/03 14:26:50 UTC
[GitHub] [kafka] yashmayya opened a new pull request, #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
yashmayya opened a new pull request, #12478:
URL: https://github.com/apache/kafka/pull/12478
- https://issues.apache.org/jira/browse/KAFKA-13952: RetryWithToleranceOperator doesn't respect infinite retries config - i.e. when `errors.retry.timeout` is set to `-1`
- From https://cwiki.apache.org/confluence/display/KAFKA/KIP-298%3A+Error+Handling+in+Connect:
>errors.retry.timeout: [-1, 0, 1, ... Long.MAX_VALUE], where -1 means infinite duration.
- Also, from the config doc: https://github.com/apache/kafka/blob/0c4da23098f8b8ae9542acd7fbaa1e5c16384a39/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java#L129-L130
- This PR fixes the issue in `RetryWithToleranceOperator` along with a couple of unit tests to ensure the same
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r960439387
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java:
##########
@@ -199,112 +202,156 @@ private RetryWithToleranceOperator setupExecutor() {
@Test
public void testExecAndHandleRetriableErrorOnce() throws Exception {
- execAndHandleRetriableError(1, 300, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 1, Collections.singletonList(300L), new RetriableException("Test"), true);
}
@Test
public void testExecAndHandleRetriableErrorThrice() throws Exception {
- execAndHandleRetriableError(3, 2100, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 3, Arrays.asList(300L, 600L, 1200L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithInfiniteRetries() throws Exception {
+ execAndHandleRetriableError(-1, 8, Arrays.asList(300L, 600L, 1200L, 2400L, 4800L, 9600L, 19200L, 38400L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithMaxRetriesExceeded() throws Exception {
+ execAndHandleRetriableError(6000, 10, Arrays.asList(300L, 600L, 1200L, 2400L, 1500L), new RetriableException("Test"), false);
}
@Test
public void testExecAndHandleNonRetriableErrorOnce() throws Exception {
- execAndHandleNonRetriableError(1, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(1, new Exception("Non Retriable Test"));
}
@Test
public void testExecAndHandleNonRetriableErrorThrice() throws Exception {
- execAndHandleNonRetriableError(3, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(3, new Exception("Non Retriable Test"));
}
- public void execAndHandleRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ @Test
+ public void testExitLatch() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(-1, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
+ EasyMock.expect(mockOperation.call()).andThrow(new RetriableException("test")).anyTimes();
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(600, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(600);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(1200, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(1200);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(2400, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(2400);
+ retryWithToleranceOperator.triggerStop();
Review Comment:
I don't think this is ideal but I couldn't think of a better place to make this call.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r960435913
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -269,7 +289,7 @@ void backoff(int attempt, long deadline) {
delay = deadline - time.milliseconds();
}
log.debug("Sleeping for {} millis", delay);
- time.sleep(delay);
+ return !exitLatch.await(delay, TimeUnit.MILLISECONDS);
Review Comment:
Ah, that's a very good point! I didn't account for this because I had (incorrectly) assumed that if a thread's interrupt status is set and a `CountDownLatch` has already been counted down to 0, a call to `CountDownLatch::await` would return immediately with `true` (i.e. the latch's count would be given "priority" over the thread's interrupt status). However, this isn't actually the case and the call to `CountDownLatch::await` will instead immediately throw an `InterruptedException` 😄
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante commented on PR #12478:
URL: https://github.com/apache/kafka/pull/12478#issuecomment-1224470640
Yeah, we can make assertions on when/how often `await` is called. If necessary, we can also manually advance the mock time by invoking `Sleep` on it in our testing logic when we're awaiting a latch, which should cause other calls to `Time::milliseconds` to reflect the waiting that we would have done on the latch.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante commented on PR #12478:
URL: https://github.com/apache/kafka/pull/12478#issuecomment-1220885110
> Why do we need to do this? If the task has been cancelled and basically abandoned by the framework, it shouldn't really matter if the backoff takes up to `errorMaxDelayInMillis` right?
We should do everything in our power to shut down the task once it's been cancelled in order to release resources allocated for the task by invoking `Task::stop`, closing the Kafka clients created for it, etc.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r955201377
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -175,9 +179,8 @@ protected <V> V execAndRetry(Operation<V> operation) throws Exception {
log.trace("Caught a retriable exception while executing {} operation with {}", context.stage(), context.executingClass());
errorHandlingMetrics.recordFailure();
if (checkRetry(startTime)) {
- backoff(attempt, deadline);
- if (Thread.currentThread().isInterrupted()) {
- log.trace("Thread was interrupted. Marking operation as failed.");
+ if (!backoff(attempt, deadline) || Thread.currentThread().isInterrupted()) {
Review Comment:
Nice catch, thanks for pointing it out! I actually wasn't sure why the interrupted status of the thread was being checked here earlier.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -269,7 +289,7 @@ void backoff(int attempt, long deadline) {
delay = deadline - time.milliseconds();
}
log.debug("Sleeping for {} millis", delay);
- time.sleep(delay);
+ return !exitLatch.await(delay, TimeUnit.MILLISECONDS);
Review Comment:
I think I get what you're saying - but why can't we simply catch `InterruptedException` here in this method itself and return `true` indicating that backoff can be called again? Then if shutdown hasn't been scheduled yet - i.e. the latch hasn't counted down to 0 yet, the next backoff will be normal; but if shutdown has been scheduled - the next call to `ShutDownLatch::await` will return immediately and `backoff()` will return false resulting in no further retries.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -331,6 +351,15 @@ public synchronized Throwable error() {
return this.context.error();
}
+ /**
+ * Exit from any currently ongoing retry loop and mark the operation as failed.
+ * This can be called from a separate thread to break out of an infinite retry loop in
+ * {@link #execAndRetry(Operation)}
+ */
+ public void exit() {
Review Comment:
Makes sense, I like both the suggested renames 👍
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -331,6 +351,15 @@ public synchronized Throwable error() {
return this.context.error();
}
+ /**
+ * Exit from any currently ongoing retry loop and mark the operation as failed.
Review Comment:
Fair point, how about - "This will stop any further retries for operations. This will also mark any ongoing operations that are currently backing off for retry as failed." ?
I think the `currently being invoked and do not complete successfully` bit might be a little tricky to interpret and should be covered by the no further retries bit anyway, WDYT?
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -175,9 +179,8 @@ protected <V> V execAndRetry(Operation<V> operation) throws Exception {
log.trace("Caught a retriable exception while executing {} operation with {}", context.stage(), context.executingClass());
errorHandlingMetrics.recordFailure();
if (checkRetry(startTime)) {
Review Comment:
Agreed. This can definitely be replaced with `time.milliseconds() < deadline` - there are some rudimentary unit tests on `checkRetry()` although I think we might be okay with losing them
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r956045554
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -175,9 +179,8 @@ protected <V> V execAndRetry(Operation<V> operation) throws Exception {
log.trace("Caught a retriable exception while executing {} operation with {}", context.stage(), context.executingClass());
errorHandlingMetrics.recordFailure();
if (checkRetry(startTime)) {
Review Comment:
Yeah, I think the lost test cases are fine as long as we still cover enough of the basic logic with more coarse-grained testing.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r963794209
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java:
##########
@@ -199,112 +202,156 @@ private RetryWithToleranceOperator setupExecutor() {
@Test
public void testExecAndHandleRetriableErrorOnce() throws Exception {
- execAndHandleRetriableError(1, 300, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 1, Collections.singletonList(300L), new RetriableException("Test"), true);
}
@Test
public void testExecAndHandleRetriableErrorThrice() throws Exception {
- execAndHandleRetriableError(3, 2100, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 3, Arrays.asList(300L, 600L, 1200L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithInfiniteRetries() throws Exception {
+ execAndHandleRetriableError(-1, 8, Arrays.asList(300L, 600L, 1200L, 2400L, 4800L, 9600L, 19200L, 38400L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithMaxRetriesExceeded() throws Exception {
+ execAndHandleRetriableError(6000, 10, Arrays.asList(300L, 600L, 1200L, 2400L, 1500L), new RetriableException("Test"), false);
}
@Test
public void testExecAndHandleNonRetriableErrorOnce() throws Exception {
- execAndHandleNonRetriableError(1, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(1, new Exception("Non Retriable Test"));
}
@Test
public void testExecAndHandleNonRetriableErrorThrice() throws Exception {
- execAndHandleNonRetriableError(3, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(3, new Exception("Non Retriable Test"));
}
- public void execAndHandleRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ @Test
+ public void testExitLatch() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(-1, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
+ EasyMock.expect(mockOperation.call()).andThrow(new RetriableException("test")).anyTimes();
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(600, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(600);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(1200, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(1200);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(2400, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(2400);
+ retryWithToleranceOperator.triggerStop();
+ return false;
+ });
- EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
- EasyMock.expect(mockOperation.call()).andReturn("Success");
-
- replay(mockOperation);
+ // expect no more calls to exitLatch.await() after retryWithToleranceOperator.triggerStop() is called
- String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertFalse(retryWithToleranceOperator.failed());
- assertEquals("Success", result);
- assertEquals(expectedWait, time.hiResClockMs());
+ exitLatch.countDown();
+ EasyMock.expectLastCall().once();
+ replay(mockOperation, exitLatch);
+ retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertEquals(4500L, time.milliseconds());
PowerMock.verifyAll();
}
- public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ public void execAndHandleRetriableError(long errorRetryTimeout, int numRetriableExceptionsThrown, List<Long> expectedWaits, Exception e, boolean successExpected) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(errorRetryTimeout, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
EasyMock.expect(mockOperation.call()).andReturn("Success");
+ for (Long expectedWait : expectedWaits) {
+ EasyMock.expect(exitLatch.await(expectedWait, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(expectedWait);
+ return false;
+ });
+ }
- replay(mockOperation);
+ replay(mockOperation, exitLatch);
String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertTrue(retryWithToleranceOperator.failed());
- assertNull(result);
- assertEquals(expectedWait, time.hiResClockMs());
+
+ if (successExpected) {
+ assertFalse(retryWithToleranceOperator.failed());
+ assertEquals("Success", result);
+ } else {
+ assertTrue(retryWithToleranceOperator.failed());
+ }
PowerMock.verifyAll();
}
- @Test
- public void testCheckRetryLimit() {
+ public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, Exception e) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(500, 100, NONE, time);
-
- time.setCurrentTimeMs(100);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
+ retryWithToleranceOperator.metrics(errorHandlingMetrics);
- time.setCurrentTimeMs(200);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
+ EasyMock.expect(mockOperation.call()).andReturn("Success");
- time.setCurrentTimeMs(400);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ // expect no call to exitLatch.await() which is only called during the retry backoff
- time.setCurrentTimeMs(499);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ replay(mockOperation, exitLatch);
- time.setCurrentTimeMs(501);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertNull(result);
- time.setCurrentTimeMs(600);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ PowerMock.verifyAll();
Review Comment:
If we don't actually invoke `mockOperation` more than once in any of the test cases with non-retriable errors, do we even need different test cases for 1 vs. 3 invocations? What's the difference between the two?
I also don't really follow why we're only verifying `mockOperation` in the success cases with retriable errors; what's preventing us from verifying it unconditionally?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r961605579
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -168,22 +170,26 @@ public synchronized <V> V execute(Operation<V> operation, Stage stage, Class<?>
* @throws Exception rethrow if a non-retriable Exception is thrown by the operation
*/
protected <V> V execAndRetry(Operation<V> operation) throws Exception {
+ RetriableException lastException = null;
int attempt = 0;
long startTime = time.milliseconds();
long deadline = (errorRetryTimeout >= 0) ? startTime + errorRetryTimeout : Long.MAX_VALUE;
do {
+ if (stopping) {
+ log.trace("Shutdown has been scheduled. Marking operation as failed.");
+ context.error(new ConnectException(lastException));
+ return null;
+ }
Review Comment:
It'd be cleaner if we put this inside the `catch` block, so that we wouldn't have to introduce the possibly-null `lastException` field.
Thinking something like this:
```java
do {
try {
attempt++;
return operation.call();
} catch (RetriableException e) {
log.trace("Caught a retriable exception while executing {} operation with {}", context.stage(), context.executingClass());
errorHandlingMetrics.recordFailure();
if (time.milliseconds() < deadline) {
backoff(attempt, deadline);
errorHandlingMetrics.recordRetry();
} else {
log.trace("Can't retry. start={}, attempt={}, deadline={}", startTime, attempt, deadline);
context.error(e);
return null;
}
if (stopping) {
log.trace("Shutdown has been scheduled. Marking operation as failed.");
context.error(e);
return null;
}
} finally {
context.attempt(attempt);
}
} while (true);
```
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java:
##########
@@ -199,112 +202,156 @@ private RetryWithToleranceOperator setupExecutor() {
@Test
public void testExecAndHandleRetriableErrorOnce() throws Exception {
- execAndHandleRetriableError(1, 300, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 1, Collections.singletonList(300L), new RetriableException("Test"), true);
}
@Test
public void testExecAndHandleRetriableErrorThrice() throws Exception {
- execAndHandleRetriableError(3, 2100, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 3, Arrays.asList(300L, 600L, 1200L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithInfiniteRetries() throws Exception {
+ execAndHandleRetriableError(-1, 8, Arrays.asList(300L, 600L, 1200L, 2400L, 4800L, 9600L, 19200L, 38400L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithMaxRetriesExceeded() throws Exception {
+ execAndHandleRetriableError(6000, 10, Arrays.asList(300L, 600L, 1200L, 2400L, 1500L), new RetriableException("Test"), false);
}
@Test
public void testExecAndHandleNonRetriableErrorOnce() throws Exception {
- execAndHandleNonRetriableError(1, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(1, new Exception("Non Retriable Test"));
}
@Test
public void testExecAndHandleNonRetriableErrorThrice() throws Exception {
- execAndHandleNonRetriableError(3, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(3, new Exception("Non Retriable Test"));
}
- public void execAndHandleRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ @Test
+ public void testExitLatch() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(-1, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
+ EasyMock.expect(mockOperation.call()).andThrow(new RetriableException("test")).anyTimes();
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(600, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(600);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(1200, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(1200);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(2400, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(2400);
+ retryWithToleranceOperator.triggerStop();
Review Comment:
I think it's fine 👍
I was wondering if we'd need to define a timeout for the test in order to prevent infinite loops, but since the exit latch is a strict mock, the test automatically fails if any unexpected waits on it occur. This is actually pretty smooth IMO
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r963777991
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -268,8 +275,12 @@ void backoff(int attempt, long deadline) {
if (delay + time.milliseconds() > deadline) {
delay = deadline - time.milliseconds();
Review Comment:
In `execAndRetry` we check to see if we've exceeded the deadline before backing off: https://github.com/yashmayya/kafka/blob/9c6dc1c3129f4166eede9f6287e44bf1a5a1bcf8/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java#L183-L184
In `backoff`, we check the current time again and, if necessary, shorten our delay in order to avoid exceeding the deadline: https://github.com/yashmayya/kafka/blob/9c6dc1c3129f4166eede9f6287e44bf1a5a1bcf8/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java#L275-L278
If we exceed the deadline after the first check in `execAndRetry` and before we shorten the delay in `backoff`, then we can end up with a negative delay right now.
We could test this by calling `MockTime::sleep` before invoking `backoff`, so that when `backoff` is called, we will have exceeded the deadline.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r965493011
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java:
##########
@@ -199,116 +202,182 @@ private RetryWithToleranceOperator setupExecutor() {
@Test
public void testExecAndHandleRetriableErrorOnce() throws Exception {
- execAndHandleRetriableError(1, 300, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 1, Collections.singletonList(300L), new RetriableException("Test"), true);
}
@Test
public void testExecAndHandleRetriableErrorThrice() throws Exception {
- execAndHandleRetriableError(3, 2100, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 3, Arrays.asList(300L, 600L, 1200L), new RetriableException("Test"), true);
}
@Test
- public void testExecAndHandleNonRetriableErrorOnce() throws Exception {
- execAndHandleNonRetriableError(1, 0, new Exception("Non Retriable Test"));
+ public void testExecAndHandleRetriableErrorWithInfiniteRetries() throws Exception {
+ execAndHandleRetriableError(-1, 8, Arrays.asList(300L, 600L, 1200L, 2400L, 4800L, 9600L, 19200L, 38400L), new RetriableException("Test"), true);
}
@Test
- public void testExecAndHandleNonRetriableErrorThrice() throws Exception {
- execAndHandleNonRetriableError(3, 0, new Exception("Non Retriable Test"));
+ public void testExecAndHandleRetriableErrorWithMaxRetriesExceeded() throws Exception {
+ execAndHandleRetriableError(6000, 6, Arrays.asList(300L, 600L, 1200L, 2400L, 1500L), new RetriableException("Test"), false);
}
- public void execAndHandleRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ public void execAndHandleRetriableError(long errorRetryTimeout, int numRetriableExceptionsThrown, List<Long> expectedWaits, Exception e, boolean successExpected) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = PowerMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(errorRetryTimeout, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
- EasyMock.expect(mockOperation.call()).andReturn("Success");
- replay(mockOperation);
+ if (successExpected) {
+ EasyMock.expect(mockOperation.call()).andReturn("Success");
+ }
+
+ for (Long expectedWait : expectedWaits) {
+ EasyMock.expect(exitLatch.await(expectedWait, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(expectedWait);
+ return false;
+ });
+ }
+
+ replay(mockOperation, exitLatch);
String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertFalse(retryWithToleranceOperator.failed());
- assertEquals("Success", result);
- assertEquals(expectedWait, time.hiResClockMs());
+ if (successExpected) {
+ assertFalse(retryWithToleranceOperator.failed());
+ assertEquals("Success", result);
+ } else {
+ assertTrue(retryWithToleranceOperator.failed());
+ }
+
+ EasyMock.verify(mockOperation);
PowerMock.verifyAll();
}
- public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ @Test
+ public void testExecAndHandleNonRetriableError() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = PowerMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
- EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
- EasyMock.expect(mockOperation.call()).andReturn("Success");
+ EasyMock.expect(mockOperation.call()).andThrow(new Exception("Test")).times(1);
- replay(mockOperation);
+ // expect no call to exitLatch.await() which is only called during the retry backoff
+
+ replay(mockOperation, exitLatch);
String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
assertTrue(retryWithToleranceOperator.failed());
assertNull(result);
- assertEquals(expectedWait, time.hiResClockMs());
+ EasyMock.verify(mockOperation);
PowerMock.verifyAll();
}
@Test
- public void testCheckRetryLimit() {
+ public void testExitLatch() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(500, 100, NONE, time);
-
- time.setCurrentTimeMs(100);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
-
- time.setCurrentTimeMs(200);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
-
- time.setCurrentTimeMs(400);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ CountDownLatch exitLatch = PowerMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(-1, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
+ retryWithToleranceOperator.metrics(errorHandlingMetrics);
+ EasyMock.expect(mockOperation.call()).andThrow(new RetriableException("test")).anyTimes();
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(600, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(600);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(1200, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(1200);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(2400, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(2400);
+ retryWithToleranceOperator.triggerStop();
+ return false;
+ });
- time.setCurrentTimeMs(499);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ // expect no more calls to exitLatch.await() after retryWithToleranceOperator.triggerStop() is called
- time.setCurrentTimeMs(501);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ exitLatch.countDown();
+ EasyMock.expectLastCall().once();
- time.setCurrentTimeMs(600);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ replay(mockOperation, exitLatch);
+ retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertEquals(4500L, time.milliseconds());
+ PowerMock.verifyAll();
}
@Test
- public void testBackoffLimit() {
+ public void testBackoffLimit() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(5, 5000, NONE, time);
+ CountDownLatch exitLatch = PowerMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(5, 5000, NONE, time, new ProcessingContext(), exitLatch);
- long prevTs = time.hiResClockMs();
- retryWithToleranceOperator.backoff(1, 5000);
- assertEquals(300, time.hiResClockMs() - prevTs);
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(600, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(600);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(1200, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(1200);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(2400, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(2400);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(500, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(500);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(0, TimeUnit.MILLISECONDS)).andReturn(false);
- prevTs = time.hiResClockMs();
- retryWithToleranceOperator.backoff(2, 5000);
- assertEquals(600, time.hiResClockMs() - prevTs);
+ replay(exitLatch);
- prevTs = time.hiResClockMs();
+ retryWithToleranceOperator.backoff(1, 5000);
+ retryWithToleranceOperator.backoff(2, 5000);
retryWithToleranceOperator.backoff(3, 5000);
- assertEquals(1200, time.hiResClockMs() - prevTs);
-
- prevTs = time.hiResClockMs();
retryWithToleranceOperator.backoff(4, 5000);
- assertEquals(2400, time.hiResClockMs() - prevTs);
-
- prevTs = time.hiResClockMs();
retryWithToleranceOperator.backoff(5, 5000);
- assertEquals(500, time.hiResClockMs() - prevTs);
-
- prevTs = time.hiResClockMs();
retryWithToleranceOperator.backoff(6, 5000);
- assertEquals(0, time.hiResClockMs() - prevTs);
PowerMock.verifyAll();
}
+ @Test
+ public void testBackoffMinDurationZero() throws Exception {
+ MockTime time = new MockTime(10, 0, 0);
+ CountDownLatch exitLatch = PowerMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(335, 5000, NONE, time, new ProcessingContext(), exitLatch);
+ retryWithToleranceOperator.metrics(errorHandlingMetrics);
+
+ EasyMock.expect(mockOperation.call()).andThrow(new RetriableException("Test")).anyTimes();
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ // Due to the auto tick setup on the MockTime instance, each call to time.milliseconds() will
+ // advance the clock by 10ms. The error retry timeout is configured to 335 ms so that after
+ // the first retry which backs off for 300 ms, just enough time will pass between the
+ // deadline check (when the current time in milliseconds will be 330) and the second backoff call
+ // so that in the backoff method, the current time (now 345) will have exceeded the deadline.
+ // This will cause the backoff delay to be 0ms (min possible delay since it shouldn't be negative)
+ EasyMock.expect(exitLatch.await(0, TimeUnit.MILLISECONDS)).andReturn(false);
+ replay(mockOperation, exitLatch);
+
+ retryWithToleranceOperator.execAndRetry(mockOperation);
+ assertTrue(retryWithToleranceOperator.failed());
+ PowerMock.verifyAll();
+ }
Review Comment:
Ah yeah this is better since the new test case is a little brittle in the sense that any changes made to the number of times `time.milliseconds()` is called in `execAndRetry` / `backoff` could potentially break the test case. It's also simpler and easier to follow. Thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on PR #12478:
URL: https://github.com/apache/kafka/pull/12478#issuecomment-1224095732
Thanks @C0urante, that makes sense. I think we can maintain a `Map<ConnectorTaskId, CountDownLatch>` in the `Worker` class (let's say `taskCancelLatch`) which gets populated in the task builder (with the `CountDownLatch` having a count of 1). The task's cancel latch could then be passed to its `RetryWithToleranceOperator` which could call `await` on it with a timeout in the backoff method instead of `Time::sleep` like you mentioned. The task cancel latch is counted down in `Worker::awaitStopTask` in the `if (!task.awaitStop(timeout))` block. However, a number of unit tests for `RetryWithToleranceOperator` rely on `MockTime` to verify whether the operator's retry and backoff mechanism is working as expected. Any ideas on how we could do this instead when using `CountDownLatch::await` with a timeout? One possible idea could be to expose a couple of retry variables for testing which would be used to determine number of retries and total retry duration, 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on PR #12478:
URL: https://github.com/apache/kafka/pull/12478#issuecomment-1240782194
Thanks for the detailed reviews and for bearing with me through multiple rounds of review on this one Chris!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante commented on PR #12478:
URL: https://github.com/apache/kafka/pull/12478#issuecomment-1220695831
I think a `Supplier<Boolean> isCancelled` would be fine to check if the operator should prematurely exit. One other thing worth accounting for is the backoff logic, which currently uses `Time::sleep` to wait in between retries. We could replace those calls with `CountDownLatch::await`, and when the task is cancelled, invoke `CountDownLatch::countDown` so that any in-progress calls to `await` return early.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r953898875
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -175,9 +179,8 @@ protected <V> V execAndRetry(Operation<V> operation) throws Exception {
log.trace("Caught a retriable exception while executing {} operation with {}", context.stage(), context.executingClass());
errorHandlingMetrics.recordFailure();
if (checkRetry(startTime)) {
- backoff(attempt, deadline);
- if (Thread.currentThread().isInterrupted()) {
- log.trace("Thread was interrupted. Marking operation as failed.");
+ if (!backoff(attempt, deadline) || Thread.currentThread().isInterrupted()) {
+ log.trace("Thread was interrupted or exit condition was triggered. Marking operation as failed.");
Review Comment:
Nit: "exit condition was triggered" is pretty vague. What about "shutdown has been scheduled"?
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -175,9 +179,8 @@ protected <V> V execAndRetry(Operation<V> operation) throws Exception {
log.trace("Caught a retriable exception while executing {} operation with {}", context.stage(), context.executingClass());
errorHandlingMetrics.recordFailure();
if (checkRetry(startTime)) {
- backoff(attempt, deadline);
- if (Thread.currentThread().isInterrupted()) {
- log.trace("Thread was interrupted. Marking operation as failed.");
+ if (!backoff(attempt, deadline) || Thread.currentThread().isInterrupted()) {
Review Comment:
We don't need the check to see if the thread is interrupted anymore; this was only included because of the semantics for `SystemTime::sleep`, which in turn used [Utils:sleep](https://github.com/apache/kafka/blob/5eff8592cc0cb57475c92bce6e80264d71a9bfbe/clients/src/main/java/org/apache/kafka/common/utils/Utils.java#L372-L383), which handled interruptions by resetting the thread interrupt flag and then returning normally.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -269,7 +289,7 @@ void backoff(int attempt, long deadline) {
delay = deadline - time.milliseconds();
}
log.debug("Sleeping for {} millis", delay);
- time.sleep(delay);
+ return !exitLatch.await(delay, TimeUnit.MILLISECONDS);
Review Comment:
I don't think we should use a return value here to indicate whether it's safe to continue or not, since it's unclear how to handle interruptions (if we get interrupted before we're scheduled for shutdown, we should keep retrying; if we're interrupted after being scheduled for shutdown, we should stop).
We can use the pattern from `AbstractWorkerSourceTask` in this class: use a latch for backing off, but also set a boolean flag when we're scheduled for shutdown, and use the latter when checking whether we should perform another retry iteration.
The control flow I have in mind is something like this:
- `backoff` tries to block the thread for the specified delay, but can return early due to interruption, shutdown, whatever. There is no return value, and it swallows `InterruptedException` instances by just returning normally
- We check the boolean flag before performing any retries and, if it's set, preemptively exit
- `exit` counts down the latch and sets the boolean flag to indicate that we should halt execution
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -253,13 +256,30 @@ public ToleranceType getErrorToleranceType() {
return errorToleranceType;
}
- // Visible for testing
+ /**
+ * Check whether we can continue retrying or not.
+ * Visible for testing.
+ * @param startTime the time in milliseconds when the retriable operation was first begun
+ * @return true if we can continue retrying; false if the retry timeout has been reached and we can't retry anymore
+ */
boolean checkRetry(long startTime) {
+ if (errorRetryTimeout < 0) {
+ return true;
+ }
return (time.milliseconds() - startTime) < errorRetryTimeout;
Review Comment:
This can be simplified:
```suggestion
return errorRetryTimeout < 0
|| (time.milliseconds() - startTime) < errorRetryTimeout;
```
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -175,9 +179,8 @@ protected <V> V execAndRetry(Operation<V> operation) throws Exception {
log.trace("Caught a retriable exception while executing {} operation with {}", context.stage(), context.executingClass());
errorHandlingMetrics.recordFailure();
if (checkRetry(startTime)) {
Review Comment:
I could be missing something, but do we even need `checkRetry`? Can we replace this with `time.milliseconds() < deadline`?
Asking because we've added special logic around `errorRetryTimeout` being negative in two different places, which seems unnecessary.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -331,6 +351,15 @@ public synchronized Throwable error() {
return this.context.error();
}
+ /**
+ * Exit from any currently ongoing retry loop and mark the operation as failed.
+ * This can be called from a separate thread to break out of an infinite retry loop in
Review Comment:
Nit: this is useful for more than just infinite retries
```suggestion
* This can be called from a separate thread to break out of retry/backoff loops in
```
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -331,6 +351,15 @@ public synchronized Throwable error() {
return this.context.error();
}
+ /**
+ * Exit from any currently ongoing retry loop and mark the operation as failed.
+ * This can be called from a separate thread to break out of an infinite retry loop in
+ * {@link #execAndRetry(Operation)}
+ */
+ public void exit() {
Review Comment:
Nit: I'm not a huge fan of `exit` since it has connotations of being able to immediately interrupt in-progress operations, which we don't provide at the moment. We could borrow from `WorkerTask` and call this `triggerStop` instead, and borrow from `AbstractWorkerSourceTask` and name the latch `stopRequestedLatch`.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -331,6 +351,15 @@ public synchronized Throwable error() {
return this.context.error();
}
+ /**
+ * Exit from any currently ongoing retry loop and mark the operation as failed.
Review Comment:
This makes it sound like we can interrupt the operation that we're retrying, but that's not the case. And because Java doesn't really provide a great way to force a thread to stop or return control to a call site, we probably can't ever really accomplish that. I do like the part about marking the operation as failed, though--nice attention to detail there 👍
What about "Attempt no further retries for any operations, and mark any ongoing operations that are blocked during backoff, or currently being invoked and do not complete successfully, as failed"?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r956044683
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -331,6 +351,15 @@ public synchronized Throwable error() {
return this.context.error();
}
+ /**
+ * Exit from any currently ongoing retry loop and mark the operation as failed.
Review Comment:
That's great! LGTM 👍
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r964923239
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java:
##########
@@ -199,116 +202,182 @@ private RetryWithToleranceOperator setupExecutor() {
@Test
public void testExecAndHandleRetriableErrorOnce() throws Exception {
- execAndHandleRetriableError(1, 300, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 1, Collections.singletonList(300L), new RetriableException("Test"), true);
}
@Test
public void testExecAndHandleRetriableErrorThrice() throws Exception {
- execAndHandleRetriableError(3, 2100, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 3, Arrays.asList(300L, 600L, 1200L), new RetriableException("Test"), true);
}
@Test
- public void testExecAndHandleNonRetriableErrorOnce() throws Exception {
- execAndHandleNonRetriableError(1, 0, new Exception("Non Retriable Test"));
+ public void testExecAndHandleRetriableErrorWithInfiniteRetries() throws Exception {
+ execAndHandleRetriableError(-1, 8, Arrays.asList(300L, 600L, 1200L, 2400L, 4800L, 9600L, 19200L, 38400L), new RetriableException("Test"), true);
}
@Test
- public void testExecAndHandleNonRetriableErrorThrice() throws Exception {
- execAndHandleNonRetriableError(3, 0, new Exception("Non Retriable Test"));
+ public void testExecAndHandleRetriableErrorWithMaxRetriesExceeded() throws Exception {
+ execAndHandleRetriableError(6000, 6, Arrays.asList(300L, 600L, 1200L, 2400L, 1500L), new RetriableException("Test"), false);
}
- public void execAndHandleRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ public void execAndHandleRetriableError(long errorRetryTimeout, int numRetriableExceptionsThrown, List<Long> expectedWaits, Exception e, boolean successExpected) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = PowerMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(errorRetryTimeout, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
- EasyMock.expect(mockOperation.call()).andReturn("Success");
- replay(mockOperation);
+ if (successExpected) {
+ EasyMock.expect(mockOperation.call()).andReturn("Success");
+ }
+
+ for (Long expectedWait : expectedWaits) {
+ EasyMock.expect(exitLatch.await(expectedWait, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(expectedWait);
+ return false;
+ });
+ }
+
+ replay(mockOperation, exitLatch);
String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertFalse(retryWithToleranceOperator.failed());
- assertEquals("Success", result);
- assertEquals(expectedWait, time.hiResClockMs());
+ if (successExpected) {
+ assertFalse(retryWithToleranceOperator.failed());
+ assertEquals("Success", result);
+ } else {
+ assertTrue(retryWithToleranceOperator.failed());
+ }
+
+ EasyMock.verify(mockOperation);
PowerMock.verifyAll();
}
- public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ @Test
+ public void testExecAndHandleNonRetriableError() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = PowerMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
- EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
- EasyMock.expect(mockOperation.call()).andReturn("Success");
+ EasyMock.expect(mockOperation.call()).andThrow(new Exception("Test")).times(1);
- replay(mockOperation);
+ // expect no call to exitLatch.await() which is only called during the retry backoff
+
+ replay(mockOperation, exitLatch);
String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
assertTrue(retryWithToleranceOperator.failed());
assertNull(result);
- assertEquals(expectedWait, time.hiResClockMs());
+ EasyMock.verify(mockOperation);
PowerMock.verifyAll();
}
@Test
- public void testCheckRetryLimit() {
+ public void testExitLatch() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(500, 100, NONE, time);
-
- time.setCurrentTimeMs(100);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
-
- time.setCurrentTimeMs(200);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
-
- time.setCurrentTimeMs(400);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ CountDownLatch exitLatch = PowerMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(-1, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
+ retryWithToleranceOperator.metrics(errorHandlingMetrics);
+ EasyMock.expect(mockOperation.call()).andThrow(new RetriableException("test")).anyTimes();
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(600, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(600);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(1200, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(1200);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(2400, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(2400);
+ retryWithToleranceOperator.triggerStop();
+ return false;
+ });
- time.setCurrentTimeMs(499);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ // expect no more calls to exitLatch.await() after retryWithToleranceOperator.triggerStop() is called
- time.setCurrentTimeMs(501);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ exitLatch.countDown();
+ EasyMock.expectLastCall().once();
- time.setCurrentTimeMs(600);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ replay(mockOperation, exitLatch);
+ retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertEquals(4500L, time.milliseconds());
+ PowerMock.verifyAll();
}
@Test
- public void testBackoffLimit() {
+ public void testBackoffLimit() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(5, 5000, NONE, time);
+ CountDownLatch exitLatch = PowerMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(5, 5000, NONE, time, new ProcessingContext(), exitLatch);
- long prevTs = time.hiResClockMs();
- retryWithToleranceOperator.backoff(1, 5000);
- assertEquals(300, time.hiResClockMs() - prevTs);
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(600, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(600);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(1200, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(1200);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(2400, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(2400);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(500, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(500);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(0, TimeUnit.MILLISECONDS)).andReturn(false);
- prevTs = time.hiResClockMs();
- retryWithToleranceOperator.backoff(2, 5000);
- assertEquals(600, time.hiResClockMs() - prevTs);
+ replay(exitLatch);
- prevTs = time.hiResClockMs();
+ retryWithToleranceOperator.backoff(1, 5000);
+ retryWithToleranceOperator.backoff(2, 5000);
retryWithToleranceOperator.backoff(3, 5000);
- assertEquals(1200, time.hiResClockMs() - prevTs);
-
- prevTs = time.hiResClockMs();
retryWithToleranceOperator.backoff(4, 5000);
- assertEquals(2400, time.hiResClockMs() - prevTs);
-
- prevTs = time.hiResClockMs();
retryWithToleranceOperator.backoff(5, 5000);
- assertEquals(500, time.hiResClockMs() - prevTs);
-
- prevTs = time.hiResClockMs();
retryWithToleranceOperator.backoff(6, 5000);
- assertEquals(0, time.hiResClockMs() - prevTs);
PowerMock.verifyAll();
}
+ @Test
+ public void testBackoffMinDurationZero() throws Exception {
+ MockTime time = new MockTime(10, 0, 0);
+ CountDownLatch exitLatch = PowerMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(335, 5000, NONE, time, new ProcessingContext(), exitLatch);
+ retryWithToleranceOperator.metrics(errorHandlingMetrics);
+
+ EasyMock.expect(mockOperation.call()).andThrow(new RetriableException("Test")).anyTimes();
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ // Due to the auto tick setup on the MockTime instance, each call to time.milliseconds() will
+ // advance the clock by 10ms. The error retry timeout is configured to 335 ms so that after
+ // the first retry which backs off for 300 ms, just enough time will pass between the
+ // deadline check (when the current time in milliseconds will be 330) and the second backoff call
+ // so that in the backoff method, the current time (now 345) will have exceeded the deadline.
+ // This will cause the backoff delay to be 0ms (min possible delay since it shouldn't be negative)
+ EasyMock.expect(exitLatch.await(0, TimeUnit.MILLISECONDS)).andReturn(false);
+ replay(mockOperation, exitLatch);
+
+ retryWithToleranceOperator.execAndRetry(mockOperation);
+ assertTrue(retryWithToleranceOperator.failed());
+ PowerMock.verifyAll();
+ }
Review Comment:
This doesn't need to be an entirely new test case; we can just tweak the existing `testBackoffLimit` case with something like this (the first and last lines are unmodified from that test case):
```java
retryWithToleranceOperator.backoff(5, 5000);
// Simulate a small delay between calculating the deadline, and backing off
time.sleep(1);
// We may try to begin backing off after the deadline has already passed; make sure
// that we don't wait with a negative timeout
retryWithToleranceOperator.backoff(6, 5000);
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on PR #12478:
URL: https://github.com/apache/kafka/pull/12478#issuecomment-1220560508
Thanks for the review @C0urante and sorry about the late reply. That's a very astute observation regarding tasks potentially getting stuck in infinite loops - you're right in that the task thread is never interrupted even if the scheduled shutdown times out.
> can you see about adding some logic to break out of retry loops here if the task is cancelled
Any pointers on how we could go about doing that? Some ideas I had in mind were around making the `WorkerTask` and its method `isCancelled()` both public, and when we instantiate `RetryWithToleranceOperator` in the `Worker` class - either configure it with its associated `WorkerTask` (i.e. there will be a circular reference since the `WorkerTask` also references the `RetryWithToleranceOperator` instance) or configure it with a method reference to the associated `WorkerTask`'s `isCancelled()` method. This way in the retry loop we can check the return value for `isCancelled()` in addition to (or instead of) the thread's interrupt status. This doesn't seem like a particularly clean solution due to the additional coupling introduced so I'd be happy to hear if you had any better ideas!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante merged pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante merged PR #12478:
URL: https://github.com/apache/kafka/pull/12478
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r962110802
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java:
##########
@@ -199,112 +202,156 @@ private RetryWithToleranceOperator setupExecutor() {
@Test
public void testExecAndHandleRetriableErrorOnce() throws Exception {
- execAndHandleRetriableError(1, 300, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 1, Collections.singletonList(300L), new RetriableException("Test"), true);
}
@Test
public void testExecAndHandleRetriableErrorThrice() throws Exception {
- execAndHandleRetriableError(3, 2100, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 3, Arrays.asList(300L, 600L, 1200L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithInfiniteRetries() throws Exception {
+ execAndHandleRetriableError(-1, 8, Arrays.asList(300L, 600L, 1200L, 2400L, 4800L, 9600L, 19200L, 38400L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithMaxRetriesExceeded() throws Exception {
+ execAndHandleRetriableError(6000, 10, Arrays.asList(300L, 600L, 1200L, 2400L, 1500L), new RetriableException("Test"), false);
}
@Test
public void testExecAndHandleNonRetriableErrorOnce() throws Exception {
- execAndHandleNonRetriableError(1, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(1, new Exception("Non Retriable Test"));
}
@Test
public void testExecAndHandleNonRetriableErrorThrice() throws Exception {
- execAndHandleNonRetriableError(3, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(3, new Exception("Non Retriable Test"));
}
- public void execAndHandleRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ @Test
+ public void testExitLatch() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(-1, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
+ EasyMock.expect(mockOperation.call()).andThrow(new RetriableException("test")).anyTimes();
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(600, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(600);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(1200, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(1200);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(2400, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(2400);
+ retryWithToleranceOperator.triggerStop();
+ return false;
+ });
- EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
- EasyMock.expect(mockOperation.call()).andReturn("Success");
-
- replay(mockOperation);
+ // expect no more calls to exitLatch.await() after retryWithToleranceOperator.triggerStop() is called
- String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertFalse(retryWithToleranceOperator.failed());
- assertEquals("Success", result);
- assertEquals(expectedWait, time.hiResClockMs());
+ exitLatch.countDown();
+ EasyMock.expectLastCall().once();
+ replay(mockOperation, exitLatch);
+ retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertEquals(4500L, time.milliseconds());
PowerMock.verifyAll();
}
- public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ public void execAndHandleRetriableError(long errorRetryTimeout, int numRetriableExceptionsThrown, List<Long> expectedWaits, Exception e, boolean successExpected) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(errorRetryTimeout, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
EasyMock.expect(mockOperation.call()).andReturn("Success");
+ for (Long expectedWait : expectedWaits) {
+ EasyMock.expect(exitLatch.await(expectedWait, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(expectedWait);
+ return false;
+ });
+ }
- replay(mockOperation);
+ replay(mockOperation, exitLatch);
String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertTrue(retryWithToleranceOperator.failed());
- assertNull(result);
- assertEquals(expectedWait, time.hiResClockMs());
+
+ if (successExpected) {
+ assertFalse(retryWithToleranceOperator.failed());
+ assertEquals("Success", result);
+ } else {
+ assertTrue(retryWithToleranceOperator.failed());
+ }
PowerMock.verifyAll();
}
- @Test
- public void testCheckRetryLimit() {
+ public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, Exception e) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(500, 100, NONE, time);
-
- time.setCurrentTimeMs(100);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
+ retryWithToleranceOperator.metrics(errorHandlingMetrics);
- time.setCurrentTimeMs(200);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
+ EasyMock.expect(mockOperation.call()).andReturn("Success");
- time.setCurrentTimeMs(400);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ // expect no call to exitLatch.await() which is only called during the retry backoff
- time.setCurrentTimeMs(499);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ replay(mockOperation, exitLatch);
- time.setCurrentTimeMs(501);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertNull(result);
- time.setCurrentTimeMs(600);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ PowerMock.verifyAll();
Review Comment:
I've added this verification to `execAndHandleRetriableError` (in the success case). But here in `execAndHandleNonRetriableError`, no matter what the value for `numNonRetriableExceptionsThrown` is, `mockOperation.call()` will only be called once (since the exception isn't retriable) - so we can't verify that the expectation that `e` is thrown `numNonRetriableExceptionsThrown` times is met. I guess this would've been easier to test if this was migrated to Mockito?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r960436353
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -175,9 +179,8 @@ protected <V> V execAndRetry(Operation<V> operation) throws Exception {
log.trace("Caught a retriable exception while executing {} operation with {}", context.stage(), context.executingClass());
errorHandlingMetrics.recordFailure();
if (checkRetry(startTime)) {
Review Comment:
Fair enough, I do think that the relevant logic is already tested by the `testExecAndHandle...` tests (I've added an additional one to check the max retries exceeded case)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on PR #12478:
URL: https://github.com/apache/kafka/pull/12478#issuecomment-1224289212
>Why construct the latches in Worker? Wouldn't it be cleaner to create them in the RetryWithToleranceOperator constructor and expose a method in that class to count down the latch, which we could then invoke from tasks inside their cancel method?
That's a fair point, I guess I was coming from the point of view of implementing a generic solution which could potentially be re-used by other operations that require signalling when a task is cancelled (i.e by awaiting on the task's cancel latch). But I think since this `RetryWithToleranceOperator` seems to be the only one where such a thing would be useful right now, I agree that your proposed solution is cleaner.
> As far as testing goes, can we just mock the latch and leave everything else basically as-is? We should continue to use the Time interface for everything except sleeping, so all the existing invocations of Time::milliseconds can stay.
The `MockTime::sleep` method advances its internal time counter which is then used in the tests to make assertions on how much time has elapsed.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r962111245
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -268,8 +275,12 @@ void backoff(int attempt, long deadline) {
if (delay + time.milliseconds() > deadline) {
delay = deadline - time.milliseconds();
Review Comment:
I guess it isn't impossible although extremely unlikely? I can add logic to set a min delay of 0 here, but I'm not sure I understand how we'd go about testing that? 🤔
Another option (which seems preferable) would be to not call `time.milliseconds()` twice - store it in a variable `currentTime` and use it instead.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r962110802
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java:
##########
@@ -199,112 +202,156 @@ private RetryWithToleranceOperator setupExecutor() {
@Test
public void testExecAndHandleRetriableErrorOnce() throws Exception {
- execAndHandleRetriableError(1, 300, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 1, Collections.singletonList(300L), new RetriableException("Test"), true);
}
@Test
public void testExecAndHandleRetriableErrorThrice() throws Exception {
- execAndHandleRetriableError(3, 2100, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 3, Arrays.asList(300L, 600L, 1200L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithInfiniteRetries() throws Exception {
+ execAndHandleRetriableError(-1, 8, Arrays.asList(300L, 600L, 1200L, 2400L, 4800L, 9600L, 19200L, 38400L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithMaxRetriesExceeded() throws Exception {
+ execAndHandleRetriableError(6000, 10, Arrays.asList(300L, 600L, 1200L, 2400L, 1500L), new RetriableException("Test"), false);
}
@Test
public void testExecAndHandleNonRetriableErrorOnce() throws Exception {
- execAndHandleNonRetriableError(1, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(1, new Exception("Non Retriable Test"));
}
@Test
public void testExecAndHandleNonRetriableErrorThrice() throws Exception {
- execAndHandleNonRetriableError(3, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(3, new Exception("Non Retriable Test"));
}
- public void execAndHandleRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ @Test
+ public void testExitLatch() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(-1, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
+ EasyMock.expect(mockOperation.call()).andThrow(new RetriableException("test")).anyTimes();
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(600, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(600);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(1200, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(1200);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(2400, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(2400);
+ retryWithToleranceOperator.triggerStop();
+ return false;
+ });
- EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
- EasyMock.expect(mockOperation.call()).andReturn("Success");
-
- replay(mockOperation);
+ // expect no more calls to exitLatch.await() after retryWithToleranceOperator.triggerStop() is called
- String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertFalse(retryWithToleranceOperator.failed());
- assertEquals("Success", result);
- assertEquals(expectedWait, time.hiResClockMs());
+ exitLatch.countDown();
+ EasyMock.expectLastCall().once();
+ replay(mockOperation, exitLatch);
+ retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertEquals(4500L, time.milliseconds());
PowerMock.verifyAll();
}
- public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ public void execAndHandleRetriableError(long errorRetryTimeout, int numRetriableExceptionsThrown, List<Long> expectedWaits, Exception e, boolean successExpected) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(errorRetryTimeout, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
EasyMock.expect(mockOperation.call()).andReturn("Success");
+ for (Long expectedWait : expectedWaits) {
+ EasyMock.expect(exitLatch.await(expectedWait, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(expectedWait);
+ return false;
+ });
+ }
- replay(mockOperation);
+ replay(mockOperation, exitLatch);
String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertTrue(retryWithToleranceOperator.failed());
- assertNull(result);
- assertEquals(expectedWait, time.hiResClockMs());
+
+ if (successExpected) {
+ assertFalse(retryWithToleranceOperator.failed());
+ assertEquals("Success", result);
+ } else {
+ assertTrue(retryWithToleranceOperator.failed());
+ }
PowerMock.verifyAll();
}
- @Test
- public void testCheckRetryLimit() {
+ public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, Exception e) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(500, 100, NONE, time);
-
- time.setCurrentTimeMs(100);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
+ retryWithToleranceOperator.metrics(errorHandlingMetrics);
- time.setCurrentTimeMs(200);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
+ EasyMock.expect(mockOperation.call()).andReturn("Success");
- time.setCurrentTimeMs(400);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ // expect no call to exitLatch.await() which is only called during the retry backoff
- time.setCurrentTimeMs(499);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ replay(mockOperation, exitLatch);
- time.setCurrentTimeMs(501);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertNull(result);
- time.setCurrentTimeMs(600);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ PowerMock.verifyAll();
Review Comment:
I've added this verification to `execAndHandleRetriableError` (in the success case). But here in `execAndHandleNonRetriableError`, no matter what the value for `numNonRetriableExceptionsThrown` is, `mockOperation.call()` will only be called once (since the exception isn't retriable) - so we can't verify that the expectation that `e` is thrown `numNonRetriableExceptionsThrown` times is met. I guess this would've been easier to test if this was migrated to Mockito?
Edit: Actually, on second thought - the test where a non-retriable exception is thrown more than once is doing basically the same thing as the test where it is only thrown once and can probably be removed.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r961727700
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -268,8 +275,12 @@ void backoff(int attempt, long deadline) {
if (delay + time.milliseconds() > deadline) {
delay = deadline - time.milliseconds();
Review Comment:
Can this ever lead to negative delays? If so, we should add a test for that case (possibly adding in a call to, e.g., `time.sleep(1)` in the test case to simulate a little bit of time passing between checks) and add logic here to set a minimum delay of zero.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java:
##########
@@ -199,112 +202,156 @@ private RetryWithToleranceOperator setupExecutor() {
@Test
public void testExecAndHandleRetriableErrorOnce() throws Exception {
- execAndHandleRetriableError(1, 300, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 1, Collections.singletonList(300L), new RetriableException("Test"), true);
}
@Test
public void testExecAndHandleRetriableErrorThrice() throws Exception {
- execAndHandleRetriableError(3, 2100, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 3, Arrays.asList(300L, 600L, 1200L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithInfiniteRetries() throws Exception {
+ execAndHandleRetriableError(-1, 8, Arrays.asList(300L, 600L, 1200L, 2400L, 4800L, 9600L, 19200L, 38400L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithMaxRetriesExceeded() throws Exception {
+ execAndHandleRetriableError(6000, 10, Arrays.asList(300L, 600L, 1200L, 2400L, 1500L), new RetriableException("Test"), false);
}
@Test
public void testExecAndHandleNonRetriableErrorOnce() throws Exception {
- execAndHandleNonRetriableError(1, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(1, new Exception("Non Retriable Test"));
}
@Test
public void testExecAndHandleNonRetriableErrorThrice() throws Exception {
- execAndHandleNonRetriableError(3, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(3, new Exception("Non Retriable Test"));
}
- public void execAndHandleRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ @Test
+ public void testExitLatch() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(-1, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
+ EasyMock.expect(mockOperation.call()).andThrow(new RetriableException("test")).anyTimes();
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(600, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(600);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(1200, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(1200);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(2400, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(2400);
+ retryWithToleranceOperator.triggerStop();
+ return false;
+ });
- EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
- EasyMock.expect(mockOperation.call()).andReturn("Success");
-
- replay(mockOperation);
+ // expect no more calls to exitLatch.await() after retryWithToleranceOperator.triggerStop() is called
- String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertFalse(retryWithToleranceOperator.failed());
- assertEquals("Success", result);
- assertEquals(expectedWait, time.hiResClockMs());
+ exitLatch.countDown();
+ EasyMock.expectLastCall().once();
+ replay(mockOperation, exitLatch);
+ retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertEquals(4500L, time.milliseconds());
PowerMock.verifyAll();
}
- public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ public void execAndHandleRetriableError(long errorRetryTimeout, int numRetriableExceptionsThrown, List<Long> expectedWaits, Exception e, boolean successExpected) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(errorRetryTimeout, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
EasyMock.expect(mockOperation.call()).andReturn("Success");
+ for (Long expectedWait : expectedWaits) {
+ EasyMock.expect(exitLatch.await(expectedWait, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(expectedWait);
+ return false;
+ });
+ }
- replay(mockOperation);
+ replay(mockOperation, exitLatch);
String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertTrue(retryWithToleranceOperator.failed());
- assertNull(result);
- assertEquals(expectedWait, time.hiResClockMs());
+
+ if (successExpected) {
+ assertFalse(retryWithToleranceOperator.failed());
+ assertEquals("Success", result);
+ } else {
+ assertTrue(retryWithToleranceOperator.failed());
+ }
PowerMock.verifyAll();
}
- @Test
- public void testCheckRetryLimit() {
+ public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, Exception e) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(500, 100, NONE, time);
-
- time.setCurrentTimeMs(100);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
+ retryWithToleranceOperator.metrics(errorHandlingMetrics);
- time.setCurrentTimeMs(200);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
+ EasyMock.expect(mockOperation.call()).andReturn("Success");
- time.setCurrentTimeMs(400);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ // expect no call to exitLatch.await() which is only called during the retry backoff
- time.setCurrentTimeMs(499);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ replay(mockOperation, exitLatch);
- time.setCurrentTimeMs(501);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertNull(result);
- time.setCurrentTimeMs(600);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ PowerMock.verifyAll();
Review Comment:
We might also want to verify the `mockOperation` (here and in other similar test cases), since that will give us guarantees about the number of times we attempted to perform the operation.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java:
##########
@@ -199,112 +202,156 @@ private RetryWithToleranceOperator setupExecutor() {
@Test
public void testExecAndHandleRetriableErrorOnce() throws Exception {
- execAndHandleRetriableError(1, 300, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 1, Collections.singletonList(300L), new RetriableException("Test"), true);
}
@Test
public void testExecAndHandleRetriableErrorThrice() throws Exception {
- execAndHandleRetriableError(3, 2100, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 3, Arrays.asList(300L, 600L, 1200L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithInfiniteRetries() throws Exception {
+ execAndHandleRetriableError(-1, 8, Arrays.asList(300L, 600L, 1200L, 2400L, 4800L, 9600L, 19200L, 38400L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithMaxRetriesExceeded() throws Exception {
+ execAndHandleRetriableError(6000, 10, Arrays.asList(300L, 600L, 1200L, 2400L, 1500L), new RetriableException("Test"), false);
}
@Test
public void testExecAndHandleNonRetriableErrorOnce() throws Exception {
- execAndHandleNonRetriableError(1, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(1, new Exception("Non Retriable Test"));
}
@Test
public void testExecAndHandleNonRetriableErrorThrice() throws Exception {
- execAndHandleNonRetriableError(3, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(3, new Exception("Non Retriable Test"));
}
- public void execAndHandleRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ @Test
+ public void testExitLatch() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(-1, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
+ EasyMock.expect(mockOperation.call()).andThrow(new RetriableException("test")).anyTimes();
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(600, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(600);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(1200, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(1200);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(2400, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(2400);
+ retryWithToleranceOperator.triggerStop();
+ return false;
+ });
- EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
- EasyMock.expect(mockOperation.call()).andReturn("Success");
-
- replay(mockOperation);
+ // expect no more calls to exitLatch.await() after retryWithToleranceOperator.triggerStop() is called
- String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertFalse(retryWithToleranceOperator.failed());
- assertEquals("Success", result);
- assertEquals(expectedWait, time.hiResClockMs());
+ exitLatch.countDown();
+ EasyMock.expectLastCall().once();
+ replay(mockOperation, exitLatch);
+ retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertEquals(4500L, time.milliseconds());
PowerMock.verifyAll();
}
- public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ public void execAndHandleRetriableError(long errorRetryTimeout, int numRetriableExceptionsThrown, List<Long> expectedWaits, Exception e, boolean successExpected) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(errorRetryTimeout, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
EasyMock.expect(mockOperation.call()).andReturn("Success");
+ for (Long expectedWait : expectedWaits) {
+ EasyMock.expect(exitLatch.await(expectedWait, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(expectedWait);
+ return false;
+ });
+ }
- replay(mockOperation);
+ replay(mockOperation, exitLatch);
String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertTrue(retryWithToleranceOperator.failed());
- assertNull(result);
- assertEquals(expectedWait, time.hiResClockMs());
+
+ if (successExpected) {
+ assertFalse(retryWithToleranceOperator.failed());
+ assertEquals("Success", result);
+ } else {
+ assertTrue(retryWithToleranceOperator.failed());
+ }
PowerMock.verifyAll();
}
- @Test
- public void testCheckRetryLimit() {
+ public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, Exception e) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(500, 100, NONE, time);
-
- time.setCurrentTimeMs(100);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
+ retryWithToleranceOperator.metrics(errorHandlingMetrics);
- time.setCurrentTimeMs(200);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
+ EasyMock.expect(mockOperation.call()).andReturn("Success");
Review Comment:
We're testing a non-retriable error; why set up this expectation?
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java:
##########
@@ -199,112 +202,156 @@ private RetryWithToleranceOperator setupExecutor() {
@Test
public void testExecAndHandleRetriableErrorOnce() throws Exception {
- execAndHandleRetriableError(1, 300, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 1, Collections.singletonList(300L), new RetriableException("Test"), true);
}
@Test
public void testExecAndHandleRetriableErrorThrice() throws Exception {
- execAndHandleRetriableError(3, 2100, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 3, Arrays.asList(300L, 600L, 1200L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithInfiniteRetries() throws Exception {
+ execAndHandleRetriableError(-1, 8, Arrays.asList(300L, 600L, 1200L, 2400L, 4800L, 9600L, 19200L, 38400L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithMaxRetriesExceeded() throws Exception {
+ execAndHandleRetriableError(6000, 10, Arrays.asList(300L, 600L, 1200L, 2400L, 1500L), new RetriableException("Test"), false);
}
@Test
public void testExecAndHandleNonRetriableErrorOnce() throws Exception {
- execAndHandleNonRetriableError(1, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(1, new Exception("Non Retriable Test"));
}
@Test
public void testExecAndHandleNonRetriableErrorThrice() throws Exception {
- execAndHandleNonRetriableError(3, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(3, new Exception("Non Retriable Test"));
}
- public void execAndHandleRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ @Test
+ public void testExitLatch() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
Review Comment:
Creating this with `EasyMock` instead of `PowerMock` skips verifications when `PowerMock.verifyAll()` is called below.
```suggestion
CountDownLatch exitLatch = PowerMock.createStrictMock(CountDownLatch.class);
```
Without this change, we don't actually ensure that `countDown` is invoked.
(The same applies wherever we create a mocked exit latch.)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r956043729
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -269,7 +289,7 @@ void backoff(int attempt, long deadline) {
delay = deadline - time.milliseconds();
}
log.debug("Sleeping for {} millis", delay);
- time.sleep(delay);
+ return !exitLatch.await(delay, TimeUnit.MILLISECONDS);
Review Comment:
With that approach, we would retry the operation one more time before invoking `backoff`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
C0urante commented on PR #12478:
URL: https://github.com/apache/kafka/pull/12478#issuecomment-1224225930
Why construct the latches in `Worker`? Wouldn't it be cleaner to create them in the `RetryWithToleranceOperator` constructor and expose a method in that class to count down the latch, which we could then invoke from tasks inside their `cancel` method?
As far as testing goes, can we just mock the latch and leave everything else basically as-is? We should continue to use the `Time` interface for everything except sleeping, so all the existing invocations of `Time::milliseconds` can stay.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r963990153
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -268,8 +275,12 @@ void backoff(int attempt, long deadline) {
if (delay + time.milliseconds() > deadline) {
delay = deadline - time.milliseconds();
Review Comment:
Ahh, I see what you're saying now. Sorry, my bad for not interpreting it correctly earlier and thanks for the detailed explanation! I've made the required changes and added a unit test using the auto tick functionality in `MockTime`
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on PR #12478:
URL: https://github.com/apache/kafka/pull/12478#issuecomment-1220870792
> I think a Supplier<Boolean> isCancelled would be fine to check if the operator should prematurely exit
Makes sense, I've implemented this with a test.
> One other thing worth accounting for is the backoff logic, which currently uses Time::sleep to wait in between retries. We could replace those calls with CountDownLatch::await, and when the task is cancelled, invoke CountDownLatch::countDown so that any in-progress calls to await return early.
Why do we need to do this? If the task has been cancelled and basically abandoned by the framework, it shouldn't really matter if the backoff takes up to `errorMaxDelayInMillis` right?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r962110842
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java:
##########
@@ -199,112 +202,156 @@ private RetryWithToleranceOperator setupExecutor() {
@Test
public void testExecAndHandleRetriableErrorOnce() throws Exception {
- execAndHandleRetriableError(1, 300, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 1, Collections.singletonList(300L), new RetriableException("Test"), true);
}
@Test
public void testExecAndHandleRetriableErrorThrice() throws Exception {
- execAndHandleRetriableError(3, 2100, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 3, Arrays.asList(300L, 600L, 1200L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithInfiniteRetries() throws Exception {
+ execAndHandleRetriableError(-1, 8, Arrays.asList(300L, 600L, 1200L, 2400L, 4800L, 9600L, 19200L, 38400L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithMaxRetriesExceeded() throws Exception {
+ execAndHandleRetriableError(6000, 10, Arrays.asList(300L, 600L, 1200L, 2400L, 1500L), new RetriableException("Test"), false);
}
@Test
public void testExecAndHandleNonRetriableErrorOnce() throws Exception {
- execAndHandleNonRetriableError(1, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(1, new Exception("Non Retriable Test"));
}
@Test
public void testExecAndHandleNonRetriableErrorThrice() throws Exception {
- execAndHandleNonRetriableError(3, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(3, new Exception("Non Retriable Test"));
}
- public void execAndHandleRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ @Test
+ public void testExitLatch() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(-1, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
+ EasyMock.expect(mockOperation.call()).andThrow(new RetriableException("test")).anyTimes();
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(600, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(600);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(1200, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(1200);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(2400, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(2400);
+ retryWithToleranceOperator.triggerStop();
+ return false;
+ });
- EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
- EasyMock.expect(mockOperation.call()).andReturn("Success");
-
- replay(mockOperation);
+ // expect no more calls to exitLatch.await() after retryWithToleranceOperator.triggerStop() is called
- String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertFalse(retryWithToleranceOperator.failed());
- assertEquals("Success", result);
- assertEquals(expectedWait, time.hiResClockMs());
+ exitLatch.countDown();
+ EasyMock.expectLastCall().once();
+ replay(mockOperation, exitLatch);
+ retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertEquals(4500L, time.milliseconds());
PowerMock.verifyAll();
}
- public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ public void execAndHandleRetriableError(long errorRetryTimeout, int numRetriableExceptionsThrown, List<Long> expectedWaits, Exception e, boolean successExpected) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(errorRetryTimeout, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
EasyMock.expect(mockOperation.call()).andReturn("Success");
+ for (Long expectedWait : expectedWaits) {
+ EasyMock.expect(exitLatch.await(expectedWait, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(expectedWait);
+ return false;
+ });
+ }
- replay(mockOperation);
+ replay(mockOperation, exitLatch);
String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertTrue(retryWithToleranceOperator.failed());
- assertNull(result);
- assertEquals(expectedWait, time.hiResClockMs());
+
+ if (successExpected) {
+ assertFalse(retryWithToleranceOperator.failed());
+ assertEquals("Success", result);
+ } else {
+ assertTrue(retryWithToleranceOperator.failed());
+ }
PowerMock.verifyAll();
}
- @Test
- public void testCheckRetryLimit() {
+ public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, Exception e) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(500, 100, NONE, time);
-
- time.setCurrentTimeMs(100);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
+ retryWithToleranceOperator.metrics(errorHandlingMetrics);
- time.setCurrentTimeMs(200);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
+ EasyMock.expect(mockOperation.call()).andReturn("Success");
Review Comment:
Hm yeah, I guess it isn't required. I'd just kept it as is from before -
https://github.com/apache/kafka/blob/8380d2edf48d3e97ebf0c2312a8127960232b947/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java#L243-L244
Removed it now, thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r962110783
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java:
##########
@@ -199,112 +202,156 @@ private RetryWithToleranceOperator setupExecutor() {
@Test
public void testExecAndHandleRetriableErrorOnce() throws Exception {
- execAndHandleRetriableError(1, 300, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 1, Collections.singletonList(300L), new RetriableException("Test"), true);
}
@Test
public void testExecAndHandleRetriableErrorThrice() throws Exception {
- execAndHandleRetriableError(3, 2100, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 3, Arrays.asList(300L, 600L, 1200L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithInfiniteRetries() throws Exception {
+ execAndHandleRetriableError(-1, 8, Arrays.asList(300L, 600L, 1200L, 2400L, 4800L, 9600L, 19200L, 38400L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithMaxRetriesExceeded() throws Exception {
+ execAndHandleRetriableError(6000, 10, Arrays.asList(300L, 600L, 1200L, 2400L, 1500L), new RetriableException("Test"), false);
}
@Test
public void testExecAndHandleNonRetriableErrorOnce() throws Exception {
- execAndHandleNonRetriableError(1, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(1, new Exception("Non Retriable Test"));
}
@Test
public void testExecAndHandleNonRetriableErrorThrice() throws Exception {
- execAndHandleNonRetriableError(3, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(3, new Exception("Non Retriable Test"));
}
- public void execAndHandleRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ @Test
+ public void testExitLatch() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
Review Comment:
Whoops, I wasn't aware of that - I haven't used EasyMock/PowerMock much. Good catch, thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r961653921
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java:
##########
@@ -168,22 +170,26 @@ public synchronized <V> V execute(Operation<V> operation, Stage stage, Class<?>
* @throws Exception rethrow if a non-retriable Exception is thrown by the operation
*/
protected <V> V execAndRetry(Operation<V> operation) throws Exception {
+ RetriableException lastException = null;
int attempt = 0;
long startTime = time.milliseconds();
long deadline = (errorRetryTimeout >= 0) ? startTime + errorRetryTimeout : Long.MAX_VALUE;
do {
+ if (stopping) {
+ log.trace("Shutdown has been scheduled. Marking operation as failed.");
+ context.error(new ConnectException(lastException));
+ return null;
+ }
Review Comment:
Fair enough, I guess we don't need to worry too much about getting stopped before the first attempt.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on code in PR #12478:
URL: https://github.com/apache/kafka/pull/12478#discussion_r963989928
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java:
##########
@@ -199,112 +202,156 @@ private RetryWithToleranceOperator setupExecutor() {
@Test
public void testExecAndHandleRetriableErrorOnce() throws Exception {
- execAndHandleRetriableError(1, 300, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 1, Collections.singletonList(300L), new RetriableException("Test"), true);
}
@Test
public void testExecAndHandleRetriableErrorThrice() throws Exception {
- execAndHandleRetriableError(3, 2100, new RetriableException("Test"));
+ execAndHandleRetriableError(6000, 3, Arrays.asList(300L, 600L, 1200L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithInfiniteRetries() throws Exception {
+ execAndHandleRetriableError(-1, 8, Arrays.asList(300L, 600L, 1200L, 2400L, 4800L, 9600L, 19200L, 38400L), new RetriableException("Test"), true);
+ }
+
+ @Test
+ public void testExecAndHandleRetriableErrorWithMaxRetriesExceeded() throws Exception {
+ execAndHandleRetriableError(6000, 10, Arrays.asList(300L, 600L, 1200L, 2400L, 1500L), new RetriableException("Test"), false);
}
@Test
public void testExecAndHandleNonRetriableErrorOnce() throws Exception {
- execAndHandleNonRetriableError(1, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(1, new Exception("Non Retriable Test"));
}
@Test
public void testExecAndHandleNonRetriableErrorThrice() throws Exception {
- execAndHandleNonRetriableError(3, 0, new Exception("Non Retriable Test"));
+ execAndHandleNonRetriableError(3, new Exception("Non Retriable Test"));
}
- public void execAndHandleRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ @Test
+ public void testExitLatch() throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(-1, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
+ EasyMock.expect(mockOperation.call()).andThrow(new RetriableException("test")).anyTimes();
+ EasyMock.expect(exitLatch.await(300, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(300);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(600, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(600);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(1200, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(1200);
+ return false;
+ });
+ EasyMock.expect(exitLatch.await(2400, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(2400);
+ retryWithToleranceOperator.triggerStop();
+ return false;
+ });
- EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
- EasyMock.expect(mockOperation.call()).andReturn("Success");
-
- replay(mockOperation);
+ // expect no more calls to exitLatch.await() after retryWithToleranceOperator.triggerStop() is called
- String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertFalse(retryWithToleranceOperator.failed());
- assertEquals("Success", result);
- assertEquals(expectedWait, time.hiResClockMs());
+ exitLatch.countDown();
+ EasyMock.expectLastCall().once();
+ replay(mockOperation, exitLatch);
+ retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertEquals(4500L, time.milliseconds());
PowerMock.verifyAll();
}
- public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, long expectedWait, Exception e) throws Exception {
+ public void execAndHandleRetriableError(long errorRetryTimeout, int numRetriableExceptionsThrown, List<Long> expectedWaits, Exception e, boolean successExpected) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time);
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(errorRetryTimeout, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
retryWithToleranceOperator.metrics(errorHandlingMetrics);
EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
EasyMock.expect(mockOperation.call()).andReturn("Success");
+ for (Long expectedWait : expectedWaits) {
+ EasyMock.expect(exitLatch.await(expectedWait, TimeUnit.MILLISECONDS)).andAnswer(() -> {
+ time.sleep(expectedWait);
+ return false;
+ });
+ }
- replay(mockOperation);
+ replay(mockOperation, exitLatch);
String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
- assertTrue(retryWithToleranceOperator.failed());
- assertNull(result);
- assertEquals(expectedWait, time.hiResClockMs());
+
+ if (successExpected) {
+ assertFalse(retryWithToleranceOperator.failed());
+ assertEquals("Success", result);
+ } else {
+ assertTrue(retryWithToleranceOperator.failed());
+ }
PowerMock.verifyAll();
}
- @Test
- public void testCheckRetryLimit() {
+ public void execAndHandleNonRetriableError(int numRetriableExceptionsThrown, Exception e) throws Exception {
MockTime time = new MockTime(0, 0, 0);
- RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(500, 100, NONE, time);
-
- time.setCurrentTimeMs(100);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ CountDownLatch exitLatch = EasyMock.createStrictMock(CountDownLatch.class);
+ RetryWithToleranceOperator retryWithToleranceOperator = new RetryWithToleranceOperator(6000, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, time, new ProcessingContext(), exitLatch);
+ retryWithToleranceOperator.metrics(errorHandlingMetrics);
- time.setCurrentTimeMs(200);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ EasyMock.expect(mockOperation.call()).andThrow(e).times(numRetriableExceptionsThrown);
+ EasyMock.expect(mockOperation.call()).andReturn("Success");
- time.setCurrentTimeMs(400);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ // expect no call to exitLatch.await() which is only called during the retry backoff
- time.setCurrentTimeMs(499);
- assertTrue(retryWithToleranceOperator.checkRetry(0));
+ replay(mockOperation, exitLatch);
- time.setCurrentTimeMs(501);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ String result = retryWithToleranceOperator.execAndHandleError(mockOperation, Exception.class);
+ assertTrue(retryWithToleranceOperator.failed());
+ assertNull(result);
- time.setCurrentTimeMs(600);
- assertFalse(retryWithToleranceOperator.checkRetry(0));
+ PowerMock.verifyAll();
Review Comment:
> do we even need different test cases for 1 vs. 3 invocations? What's the difference between the two?
There isn't any difference between the test cases for 1 vs 3 invocations for non-retriable errors, that's why I suggested consolidating them into a single test case in my previous comment (I've made this change now).
> I also don't really follow why we're only verifying mockOperation in the success cases with retriable errors; what's preventing us from verifying it unconditionally?
Nothing actually - I've updated the test.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on pull request #12478: KAFKA-13952 fix RetryWithToleranceOperator to respect infinite retries configuration
Posted by GitBox <gi...@apache.org>.
yashmayya commented on PR #12478:
URL: https://github.com/apache/kafka/pull/12478#issuecomment-1207201127
@C0urante @tombentley could you please take a look at this simple bug fix?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org