KarmaGYZ commented on a change in pull request #13311:
URL: https://github.com/apache/flink/pull/13311#discussion_r488345943
##########
File path:
flink-yarn/src/test/java/org/apache/flink/yarn/TestingYarnNMClientAsync.java
##########
@@ -51,30 +65,74 @@ public void stopContainerAsync(ContainerId containerId,
NodeId nodeId) {
this.stopContainerAsyncConsumer.accept(containerId, nodeId,
callbackHandler);
}
- void setStartContainerAsyncConsumer(TriConsumer<Container,
ContainerLaunchContext, CallbackHandler> startContainerAsyncConsumer) {
- this.startContainerAsyncConsumer =
Preconditions.checkNotNull(startContainerAsyncConsumer);
- }
-
- void setStopContainerAsyncConsumer(TriConsumer<ContainerId, NodeId,
CallbackHandler> stopContainerAsyncConsumer) {
- this.stopContainerAsyncConsumer =
Preconditions.checkNotNull(stopContainerAsyncConsumer);
+ static Builder builder() {
+ return new Builder();
}
//
------------------------------------------------------------------------
// Override lifecycle methods to avoid actually starting the service
//
------------------------------------------------------------------------
@Override
- protected void serviceInit(Configuration conf) throws Exception {
- // noop
+ public void init(Configuration conf) {
+ clientInitRunnable.run();
}
@Override
- protected void serviceStart() throws Exception {
- // noop
+ public void start() {
+ clientStartRunnable.run();
}
@Override
- protected void serviceStop() throws Exception {
- // noop
+ public void stop() {
+ clientStopRunnable.run();
+ }
+
+ /**
+ * Builder class for {@link TestingYarnAMRMClientAsync}.
+ */
+ public static class Builder {
+ private volatile TriConsumer<Container, ContainerLaunchContext,
CallbackHandler> startContainerAsyncConsumer = (ignored1, ignored2, ignored3)
-> {};
+ private volatile TriConsumer<ContainerId, NodeId,
CallbackHandler> stopContainerAsyncConsumer = (ignored1, ignored2, ignored3) ->
{};
+ private volatile Runnable clientInitRunnable = () -> {};
+ private volatile Runnable clientStartRunnable = () -> {};
+ private volatile Runnable clientStopRunnable = () -> {};
Review comment:
Ditto.
##########
File path:
flink-yarn/src/test/java/org/apache/flink/yarn/TestingYarnAMRMClientAsync.java
##########
@@ -45,25 +45,40 @@
*/
public class TestingYarnAMRMClientAsync extends
AMRMClientAsyncImpl<AMRMClient.ContainerRequest> {
- private volatile Function<Tuple4<Priority, String, Resource,
CallbackHandler>, List<? extends Collection<AMRMClient.ContainerRequest>>>
- getMatchingRequestsFunction = ignored ->
Collections.emptyList();
- private volatile BiConsumer<AMRMClient.ContainerRequest,
CallbackHandler> addContainerRequestConsumer = (ignored1, ignored2) -> {};
- private volatile BiConsumer<AMRMClient.ContainerRequest,
CallbackHandler> removeContainerRequestConsumer = (ignored1, ignored2) -> {};
- private volatile BiConsumer<ContainerId, CallbackHandler>
releaseAssignedContainerConsumer = (ignored1, ignored2) -> {};
- private volatile Consumer<Integer> setHeartbeatIntervalConsumer =
(ignored) -> {};
- private volatile TriFunction<String, Integer, String,
RegisterApplicationMasterResponse> registerApplicationMasterFunction =
- (ignored1, ignored2, ignored3) ->
RegisterApplicationMasterResponse.newInstance(
- Resource.newInstance(0, 0),
- Resource.newInstance(Integer.MAX_VALUE,
Integer.MAX_VALUE),
- Collections.emptyMap(),
- null,
- Collections.emptyList(),
- null,
- Collections.emptyList());
- private volatile TriConsumer<FinalApplicationStatus, String, String>
unregisterApplicationMasterConsumer = (ignored1, ignored2, ignored3) -> {};
-
- TestingYarnAMRMClientAsync(CallbackHandler callbackHandler) {
+ private volatile Function<Tuple4<Priority, String, Resource,
CallbackHandler>, List<? extends Collection<AMRMClient.ContainerRequest>>>
getMatchingRequestsFunction;
+ private volatile BiConsumer<AMRMClient.ContainerRequest,
CallbackHandler> addContainerRequestConsumer;
+ private volatile BiConsumer<AMRMClient.ContainerRequest,
CallbackHandler> removeContainerRequestConsumer;
+ private volatile BiConsumer<ContainerId, CallbackHandler>
releaseAssignedContainerConsumer;
+ private volatile Consumer<Integer> setHeartbeatIntervalConsumer;
+ private volatile TriFunction<String, Integer, String,
RegisterApplicationMasterResponse> registerApplicationMasterFunction;
+ private volatile TriConsumer<FinalApplicationStatus, String, String>
unregisterApplicationMasterConsumer;
+ private volatile Runnable clientInitRunnable;
+ private volatile Runnable clientStartRunnable;
+ private volatile Runnable clientStopRunnable;
Review comment:
Ditto.
##########
File path:
flink-yarn/src/test/java/org/apache/flink/yarn/TestingYarnAMRMClientAsync.java
##########
@@ -101,58 +116,111 @@ public void
unregisterApplicationMaster(FinalApplicationStatus appStatus, String
unregisterApplicationMasterConsumer.accept(appStatus,
appMessage, appTrackingUrl);
}
- void setGetMatchingRequestsFunction(
- Function<Tuple4<Priority, String, Resource, CallbackHandler>,
List<? extends Collection<AMRMClient.ContainerRequest>>>
- getMatchingRequestsFunction) {
- this.getMatchingRequestsFunction =
Preconditions.checkNotNull(getMatchingRequestsFunction);
- }
-
- void setAddContainerRequestConsumer(
- BiConsumer<AMRMClient.ContainerRequest, CallbackHandler>
addContainerRequestConsumer) {
- this.addContainerRequestConsumer =
Preconditions.checkNotNull(addContainerRequestConsumer);
- }
-
- void setRemoveContainerRequestConsumer(
- BiConsumer<AMRMClient.ContainerRequest, CallbackHandler>
removeContainerRequestConsumer) {
- this.removeContainerRequestConsumer =
Preconditions.checkNotNull(removeContainerRequestConsumer);
- }
-
- void setReleaseAssignedContainerConsumer(
- BiConsumer<ContainerId, CallbackHandler>
releaseAssignedContainerConsumer) {
- this.releaseAssignedContainerConsumer =
Preconditions.checkNotNull(releaseAssignedContainerConsumer);
- }
-
- void setSetHeartbeatIntervalConsumer(
- Consumer<Integer> setHeartbeatIntervalConsumer) {
- this.setHeartbeatIntervalConsumer =
setHeartbeatIntervalConsumer;
- }
-
- void setRegisterApplicationMasterFunction(
- TriFunction<String, Integer, String,
RegisterApplicationMasterResponse> registerApplicationMasterFunction) {
- this.registerApplicationMasterFunction =
registerApplicationMasterFunction;
- }
-
- void setUnregisterApplicationMasterConsumer(
- TriConsumer<FinalApplicationStatus, String, String>
unregisterApplicationMasterConsumer) {
- this.unregisterApplicationMasterConsumer =
unregisterApplicationMasterConsumer;
+ static Builder builder() {
+ return new Builder();
}
//
------------------------------------------------------------------------
// Override lifecycle methods to avoid actually starting the service
//
------------------------------------------------------------------------
@Override
- protected void serviceInit(Configuration conf) throws Exception {
- // noop
+ public void init(Configuration conf) {
+ clientInitRunnable.run();
}
@Override
- protected void serviceStart() throws Exception {
- // noop
+ public void start() {
+ clientStartRunnable.run();
}
@Override
- protected void serviceStop() throws Exception {
- // noop
+ public void stop() {
+ clientStopRunnable.run();
+ }
+
+ /**
+ * Builder class for {@link TestingYarnAMRMClientAsync}.
+ */
+ public static class Builder {
+ private volatile Function<Tuple4<Priority, String, Resource,
CallbackHandler>, List<? extends Collection<AMRMClient.ContainerRequest>>>
+ getMatchingRequestsFunction = ignored ->
Collections.emptyList();
+ private volatile BiConsumer<AMRMClient.ContainerRequest,
CallbackHandler> addContainerRequestConsumer = (ignored1, ignored2) -> {};
+ private volatile BiConsumer<AMRMClient.ContainerRequest,
CallbackHandler> removeContainerRequestConsumer = (ignored1, ignored2) -> {};
+ private volatile BiConsumer<ContainerId, CallbackHandler>
releaseAssignedContainerConsumer = (ignored1, ignored2) -> {};
+ private volatile Consumer<Integer> setHeartbeatIntervalConsumer
= (ignored) -> {};
+ private volatile TriFunction<String, Integer, String,
RegisterApplicationMasterResponse> registerApplicationMasterFunction =
+ (ignored1, ignored2, ignored3) -> new
TestingRegisterApplicationMasterResponse(Collections::emptyList);
+ private volatile TriConsumer<FinalApplicationStatus, String,
String> unregisterApplicationMasterConsumer = (ignored1, ignored2, ignored3) ->
{};
+ private volatile Runnable clientInitRunnable = () -> {};
+ private volatile Runnable clientStartRunnable = () -> {};
+ private volatile Runnable clientStopRunnable = () -> {};
Review comment:
Ditto.
----------------------------------------------------------------
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:
[email protected]
