C0urante commented on code in PR #12823:
URL: https://github.com/apache/kafka/pull/12823#discussion_r1015522678
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitterTest.java:
##########
@@ -185,8 +160,13 @@ public void testRemove() throws Exception {
} catch (ConnectException e) {
//ignore
}
+ }
- PowerMock.verifyAll();
+ private void removeSetup() {
Review Comment:
Nit: "remove setup" kind of sounds like "remove" is a verb and we're
removing something. Maybe `expectRemove`, which would also line up nicely with
other utility functions in other tests?
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitterTest.java:
##########
@@ -71,122 +74,74 @@ public void setup() {
Long.toString(DEFAULT_OFFSET_COMMIT_INTERVAL_MS));
WorkerConfig config = new StandaloneConfig(workerProps);
committer = new SourceTaskOffsetCommitter(config, executor,
committers);
- Whitebox.setInternalState(SourceTaskOffsetCommitter.class, "log",
mockLog);
}
@SuppressWarnings("unchecked")
@Test
public void testSchedule() {
- Capture<Runnable> taskWrapper = EasyMock.newCapture();
+ ArgumentCaptor<Runnable> taskWrapper =
ArgumentCaptor.forClass(Runnable.class);
- EasyMock.expect(executor.scheduleWithFixedDelay(
- EasyMock.capture(taskWrapper),
eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS),
+ when(executor.scheduleWithFixedDelay(
+ taskWrapper.capture(), eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS),
eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS),
eq(TimeUnit.MILLISECONDS))
- ).andReturn((ScheduledFuture) commitFuture);
-
- PowerMock.replayAll();
+ ).thenReturn((ScheduledFuture) commitFuture);
committer.schedule(taskId, task);
- assertTrue(taskWrapper.hasCaptured());
assertNotNull(taskWrapper.getValue());
assertEquals(singletonMap(taskId, commitFuture), committers);
-
- PowerMock.verifyAll();
}
@Test
public void testClose() throws Exception {
long timeoutMs = 1000;
// Normal termination, where termination times out.
- executor.shutdown();
- PowerMock.expectLastCall();
-
- EasyMock.expect(executor.awaitTermination(eq(timeoutMs),
eq(TimeUnit.MILLISECONDS)))
- .andReturn(false);
- mockLog.error(EasyMock.anyString());
- PowerMock.expectLastCall();
- PowerMock.replayAll();
+ when(executor.awaitTermination(timeoutMs,
TimeUnit.MILLISECONDS)).thenReturn(false);
committer.close(timeoutMs);
- PowerMock.verifyAll();
- PowerMock.resetAll();
-
// Termination interrupted
- executor.shutdown();
- PowerMock.expectLastCall();
-
- EasyMock.expect(executor.awaitTermination(eq(timeoutMs),
eq(TimeUnit.MILLISECONDS)))
- .andThrow(new InterruptedException());
- PowerMock.replayAll();
+ when(executor.awaitTermination(timeoutMs,
TimeUnit.MILLISECONDS)).thenThrow(new InterruptedException());
committer.close(timeoutMs);
- PowerMock.verifyAll();
+ verify(executor, times(2)).shutdown();
}
@Test
public void testRemove() throws Exception {
Review Comment:
Looks great!
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitterTest.java:
##########
@@ -71,112 +75,83 @@ public void setup() {
Long.toString(DEFAULT_OFFSET_COMMIT_INTERVAL_MS));
WorkerConfig config = new StandaloneConfig(workerProps);
committer = new SourceTaskOffsetCommitter(config, executor,
committers);
- Whitebox.setInternalState(SourceTaskOffsetCommitter.class, "log",
mockLog);
}
@SuppressWarnings("unchecked")
@Test
public void testSchedule() {
- Capture<Runnable> taskWrapper = EasyMock.newCapture();
+ ArgumentCaptor<Runnable> taskWrapper =
ArgumentCaptor.forClass(Runnable.class);
- EasyMock.expect(executor.scheduleWithFixedDelay(
- EasyMock.capture(taskWrapper),
eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS),
+ when(executor.scheduleWithFixedDelay(
+ taskWrapper.capture(), eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS),
eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS),
eq(TimeUnit.MILLISECONDS))
- ).andReturn((ScheduledFuture) commitFuture);
-
- PowerMock.replayAll();
+ ).thenReturn((ScheduledFuture) commitFuture);
committer.schedule(taskId, task);
- assertTrue(taskWrapper.hasCaptured());
assertNotNull(taskWrapper.getValue());
assertEquals(singletonMap(taskId, commitFuture), committers);
-
- PowerMock.verifyAll();
}
@Test
- public void testClose() throws Exception {
+ public void testCloseWithinTimeout() throws Exception {
Review Comment:
Nit: this name seems to go against the case that we're covering here (where
`close` fails to complete within the expected timeout). Maybe rename to
`testCloseTimeout`?
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitterTest.java:
##########
@@ -185,8 +160,13 @@ public void testRemove() throws Exception {
} catch (ConnectException e) {
//ignore
}
Review Comment:
We can replace this with `assertThrows`:
```java
committers.put(taskId, taskFuture);
assertThrows(ConnectException.class, () -> committer.remove(taskId));
```
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitterTest.java:
##########
@@ -71,112 +75,83 @@ public void setup() {
Long.toString(DEFAULT_OFFSET_COMMIT_INTERVAL_MS));
WorkerConfig config = new StandaloneConfig(workerProps);
committer = new SourceTaskOffsetCommitter(config, executor,
committers);
- Whitebox.setInternalState(SourceTaskOffsetCommitter.class, "log",
mockLog);
}
@SuppressWarnings("unchecked")
@Test
public void testSchedule() {
- Capture<Runnable> taskWrapper = EasyMock.newCapture();
+ ArgumentCaptor<Runnable> taskWrapper =
ArgumentCaptor.forClass(Runnable.class);
- EasyMock.expect(executor.scheduleWithFixedDelay(
- EasyMock.capture(taskWrapper),
eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS),
+ when(executor.scheduleWithFixedDelay(
+ taskWrapper.capture(), eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS),
eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS),
eq(TimeUnit.MILLISECONDS))
- ).andReturn((ScheduledFuture) commitFuture);
-
- PowerMock.replayAll();
+ ).thenReturn((ScheduledFuture) commitFuture);
committer.schedule(taskId, task);
- assertTrue(taskWrapper.hasCaptured());
assertNotNull(taskWrapper.getValue());
assertEquals(singletonMap(taskId, commitFuture), committers);
-
- PowerMock.verifyAll();
}
@Test
- public void testClose() throws Exception {
+ public void testCloseWithinTimeout() throws Exception {
long timeoutMs = 1000;
// Normal termination, where termination times out.
- executor.shutdown();
- PowerMock.expectLastCall();
+ when(executor.awaitTermination(timeoutMs,
TimeUnit.MILLISECONDS)).thenReturn(false);
- EasyMock.expect(executor.awaitTermination(eq(timeoutMs),
eq(TimeUnit.MILLISECONDS)))
- .andReturn(false);
- mockLog.error(EasyMock.anyString());
- PowerMock.expectLastCall();
- PowerMock.replayAll();
+ try (LogCaptureAppender logCaptureAppender =
LogCaptureAppender.createAndRegister(SourceTaskOffsetCommitter.class)) {
+ committer.close(timeoutMs);
+ assertTrue(logCaptureAppender.getEvents().stream().anyMatch(e ->
e.getLevel().equals("ERROR")));
+ }
- committer.close(timeoutMs);
+ verify(executor).shutdown();
+ }
- PowerMock.verifyAll();
- PowerMock.resetAll();
+ @Test
+ public void testCloseOutsideOfTimeout() throws InterruptedException {
Review Comment:
Nit: "outside of timeout" is a little unclear; maybe just
`testCloseInterrupted`?
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
