XComp commented on a change in pull request #19116:
URL: https://github.com/apache/flink/pull/19116#discussion_r836421813
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingRetrievalBase.java
##########
@@ -45,51 +42,36 @@
private String oldAddress;
private Throwable error;
- public String waitForNewLeader(long timeout) throws Exception {
+ public String waitForNewLeader() throws Exception {
throwExceptionIfNotNull();
- final String errorMsg =
- "Listener was not notified about a new leader within " +
timeout + "ms";
CommonTestUtils.waitUntilCondition(
() -> {
- leader = leaderEventQueue.poll(timeout,
TimeUnit.MILLISECONDS);
+ leader = leaderEventQueue.take();
return leader != null
&& !leader.isEmpty()
Review comment:
```suggestion
return !leader.isEmpty()
```
null check obsolete
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderBase.java
##########
@@ -41,47 +38,32 @@
private boolean isLeader = false;
private Throwable error;
- public void waitForLeader(long timeout) throws Exception {
+ public void waitForLeader() throws Exception {
throwExceptionIfNotNull();
- final String errorMsg = "Contender was not elected as the leader
within " + timeout + "ms";
CommonTestUtils.waitUntilCondition(
() -> {
- final LeaderInformation leader =
- leaderEventQueue.poll(timeout,
TimeUnit.MILLISECONDS);
+ final LeaderInformation leader = leaderEventQueue.take();
return leader != null && !leader.isEmpty();
- },
- Deadline.fromNow(Duration.ofMillis(timeout)),
- errorMsg);
+ });
isLeader = true;
}
- public void waitForRevokeLeader(long timeout) throws Exception {
+ public void waitForRevokeLeader() throws Exception {
throwExceptionIfNotNull();
- final String errorMsg = "Contender was not revoked within " + timeout
+ "ms";
CommonTestUtils.waitUntilCondition(
() -> {
- final LeaderInformation leader =
- leaderEventQueue.poll(timeout,
TimeUnit.MILLISECONDS);
+ final LeaderInformation leader = leaderEventQueue.take();
return leader != null && leader.isEmpty();
Review comment:
```suggestion
return !leader.isEmpty();
```
The `null` check is not necessary anymore.
##########
File path:
flink-clients/src/test/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrapITCase.java
##########
@@ -223,7 +212,6 @@ public JobResultStore getJobResultStore() {
@Test
public void testSubmitFailedJobOnApplicationError() throws Exception {
- final Deadline deadline = Deadline.fromNow(TIMEOUT);
Review comment:
unused variable `TIMEOUT` in `ApplicationDispatcherBootstrapITCase`
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorITCase.java
##########
@@ -137,9 +136,7 @@ public void testJobRecoveryWithFailingTaskExecutor() throws
Exception {
assertThat(jobResultFuture.isDone(), is(false));
CommonTestUtils.waitUntilCondition(
- jobIsRunning(() ->
miniCluster.getExecutionGraph(jobGraph.getJobID())),
- Deadline.fromNow(TESTING_TIMEOUT),
Review comment:
The unused member `TESTING_TIMEOUT` can be removed...
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderBase.java
##########
@@ -41,47 +38,32 @@
private boolean isLeader = false;
private Throwable error;
- public void waitForLeader(long timeout) throws Exception {
+ public void waitForLeader() throws Exception {
throwExceptionIfNotNull();
- final String errorMsg = "Contender was not elected as the leader
within " + timeout + "ms";
CommonTestUtils.waitUntilCondition(
() -> {
- final LeaderInformation leader =
- leaderEventQueue.poll(timeout,
TimeUnit.MILLISECONDS);
+ final LeaderInformation leader = leaderEventQueue.take();
return leader != null && !leader.isEmpty();
Review comment:
```suggestion
return !leader.isEmpty();
```
The `null` check is not necessary anymore.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/testutils/MiniClusterResource.java
##########
@@ -159,8 +159,7 @@ private void cancelAllJobs(boolean waitUntilSlotsAreFreed) {
.isGloballyTerminalState())
.count();
return unfinishedJobs == 0;
- },
- jobCancellationDeadline);
Review comment:
It feels like we should get rid of the `jobCancellationDeadline`
entirely in this method. There's no point in waiting forever for the jobs to
finish if there's already another call that is actually blocked but runs into a
timeout further up n this method...
##########
File path:
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/TestingLeaderCallbackHandler.java
##########
@@ -70,31 +68,23 @@ public static String waitUntilNewLeaderAppears(long
timeout) throws Exception {
final String lockIdentity = sharedQueue.poll(timeout,
TimeUnit.MILLISECONDS);
Review comment:
I guess we could simplify the method even more (removing the `timeout`
parameter) and using `sharedQueue.take()` instead of `poll`. WDYT?
##########
File path:
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/TestingLeaderCallbackHandler.java
##########
@@ -70,31 +68,23 @@ public static String waitUntilNewLeaderAppears(long
timeout) throws Exception {
final String lockIdentity = sharedQueue.poll(timeout,
TimeUnit.MILLISECONDS);
leaderRef.set(lockIdentity);
return lockIdentity != null;
- },
- Deadline.fromNow(Duration.ofMillis(timeout)),
- "No leader is elected with " + timeout + "ms");
+ });
return leaderRef.get();
}
public void waitForNewLeader(long timeout) throws Exception {
- final String errorMsg =
- "No leader with " + lockIdentity + " is elected within " +
timeout + "ms";
- poll(leaderQueue, timeout, errorMsg);
+ poll(leaderQueue, timeout);
}
public void waitForRevokeLeader(long timeout) throws Exception {
- final String errorMsg =
- "No leader with " + lockIdentity + " is revoke within " +
timeout + "ms";
- poll(revokeQueue, timeout, errorMsg);
+ poll(revokeQueue, timeout);
}
- private void poll(BlockingQueue<String> queue, long timeout, String
errorMsg) throws Exception {
+ private void poll(BlockingQueue<String> queue, long timeout) throws
Exception {
CommonTestUtils.waitUntilCondition(
() -> {
final String lockIdentity = queue.poll(timeout,
TimeUnit.MILLISECONDS);
Review comment:
Same here, `take` instead of `poll` would enablue us to remove the
`timeout` parameter from this and its calling methods as well
##########
File path:
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesStateHandleStoreTest.java
##########
@@ -445,7 +445,7 @@ public void testReplaceWithNoLeadershipAndDiscardState()
throws Exception {
store.addAndLock(key, state);
// Lost leadership
getLeaderCallback().notLeader();
- electionEventHandler.waitForRevokeLeader(TIMEOUT);
Review comment:
`TIMEOUT` in parent class could be annotated as `private` now
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingRetrievalBase.java
##########
@@ -45,51 +42,36 @@
private String oldAddress;
private Throwable error;
- public String waitForNewLeader(long timeout) throws Exception {
+ public String waitForNewLeader() throws Exception {
throwExceptionIfNotNull();
- final String errorMsg =
- "Listener was not notified about a new leader within " +
timeout + "ms";
CommonTestUtils.waitUntilCondition(
() -> {
- leader = leaderEventQueue.poll(timeout,
TimeUnit.MILLISECONDS);
+ leader = leaderEventQueue.take();
return leader != null
&& !leader.isEmpty()
&& !leader.getLeaderAddress().equals(oldAddress);
- },
- Deadline.fromNow(Duration.ofMillis(timeout)),
- errorMsg);
+ });
oldAddress = leader.getLeaderAddress();
return leader.getLeaderAddress();
}
- public void waitForEmptyLeaderInformation(long timeout) throws Exception {
+ public void waitForEmptyLeaderInformation() throws Exception {
throwExceptionIfNotNull();
- final String errorMsg =
- "Listener was not notified about an empty leader within " +
timeout + "ms";
CommonTestUtils.waitUntilCondition(
() -> {
- leader = leaderEventQueue.poll(timeout,
TimeUnit.MILLISECONDS);
+ leader = leaderEventQueue.take();
return leader != null && leader.isEmpty();
Review comment:
```suggestion
return leader.isEmpty();
```
null check obsolete
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
