tillrohrmann commented on a change in pull request #13644: URL: https://github.com/apache/flink/pull/13644#discussion_r513355235
########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderBase.java ########## @@ -0,0 +1,116 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.leaderelection; + +import java.util.concurrent.TimeoutException; +import java.util.function.Supplier; + +/** + * Base class which provides some convenience functions for testing purposes of {@link LeaderContender} and + * {@link LeaderElectionEventHandler}. + */ +public class TestingLeaderBase { + + protected boolean leader = false; + protected Throwable error = null; + + protected final Object lock = new Object(); + private final Object errorLock = new Object(); + + /** + * Waits until the contender becomes the leader or until the timeout has been exceeded. + * + * @param timeout + * @throws TimeoutException + */ + public void waitForLeader(long timeout) throws TimeoutException { + waitFor(this::isLeader, timeout, "Contender was not elected as the leader within " + timeout + "ms"); + } + + /** + * Waits until the contender revokes the leader or until the timeout has been exceeded. + * + * @param timeout + * @throws TimeoutException + */ + public void waitForRevokeLeader(long timeout) throws TimeoutException { + waitFor(() -> !isLeader(), timeout, "Contender was not revoked within " + timeout + "ms"); + } + + protected void waitFor(Supplier<Boolean> supplier, long timeout, String msg) throws TimeoutException { + long start = System.currentTimeMillis(); + long curTimeout; + + while (!supplier.get() && (curTimeout = timeout - System.currentTimeMillis() + start) > 0) { + synchronized (lock) { + try { + lock.wait(curTimeout); + } catch (InterruptedException e) { + // we got interrupted so check again for the condition + } + } + } + + if (!supplier.get()) { + throw new TimeoutException(msg); + } Review comment: This is not strictly needed right now. We could tackle this as part of a follow up issue. ########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingRetrievalBase.java ########## @@ -0,0 +1,95 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.leaderelection; + +import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalListener; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.UUID; +import java.util.concurrent.TimeoutException; + +/** + * Base class which provides some convenience functions for testing purposes of {@link LeaderRetrievalListener} and + * {@link org.apache.flink.runtime.leaderretrieval.LeaderRetrievalEventHandler}. + */ +public class TestingRetrievalBase { + + protected final Logger logger = LoggerFactory.getLogger(getClass()); + private final Object lock = new Object(); + + private String address; + private String oldAddress; + private UUID leaderSessionID; + private Exception exception; + + + public String getAddress() { + return address; + } + + public UUID getLeaderSessionID() { + return leaderSessionID; + } + + public String waitForNewLeader(long timeout) throws Exception { + long start = System.currentTimeMillis(); + long curTimeout; + + synchronized (lock) { + while ( + exception == null && + (address == null || address.equals(oldAddress)) && + (curTimeout = timeout - System.currentTimeMillis() + start) > 0) { + try { + lock.wait(curTimeout); + } catch (InterruptedException e) { + // we got interrupted so check again for the condition + } + } + } Review comment: Same here. This can be a follow up task. ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
