azagrebin commented on a change in pull request #7526: [FLINK-11353][tests]
Port JobManagerHAJobGraphRecoveryITCase to new code base
URL: https://github.com/apache/flink/pull/7526#discussion_r250231423
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherHATest.java
##########
@@ -225,31 +206,82 @@ public static JobGraph createNonEmptyJobGraph() {
return jobGraph;
}
- private static class HATestingDispatcher extends TestingDispatcher {
+ @Nonnull
+ private HATestingDispatcher
createDispatcher(TestingHighAvailabilityServices highAvailabilityServices,
Queue<DispatcherId> fencingTokens) throws Exception {
+ return createDispatcher(highAvailabilityServices,
fencingTokens, new TestingJobManagerRunnerFactory(new CompletableFuture<>(),
new CompletableFuture<>(), CompletableFuture.completedFuture(null)));
+ }
- @Nonnull
- private final BlockingQueue<DispatcherId> fencingTokens;
+ @Nonnull
+ private HATestingDispatcher createDispatcher(
+ TestingHighAvailabilityServices highAvailabilityServices,
+ @Nullable Queue<DispatcherId> fencingTokens,
+ Dispatcher.JobManagerRunnerFactory jobManagerRunnerFactory)
throws Exception {
+ final Configuration configuration = new Configuration();
- HATestingDispatcher(RpcService rpcService, String endpointId,
Configuration configuration, HighAvailabilityServices highAvailabilityServices,
ResourceManagerGateway resourceManagerGateway, BlobServer blobServer,
HeartbeatServices heartbeatServices, JobManagerMetricGroup
jobManagerMetricGroup, @Nullable String metricQueryServicePath,
ArchivedExecutionGraphStore archivedExecutionGraphStore,
JobManagerRunnerFactory jobManagerRunnerFactory, FatalErrorHandler
fatalErrorHandler, @Nonnull BlockingQueue<DispatcherId> fencingTokens) throws
Exception {
- super(rpcService, endpointId, configuration,
highAvailabilityServices, resourceManagerGateway, blobServer,
heartbeatServices, jobManagerMetricGroup, metricQueryServicePath,
archivedExecutionGraphStore, jobManagerRunnerFactory, fatalErrorHandler);
- this.fencingTokens = fencingTokens;
- }
+ return new HATestingDispatcher(
+ rpcService,
+ UUID.randomUUID().toString(),
+ configuration,
+ highAvailabilityServices,
+ new TestingResourceManagerGateway(),
+ new BlobServer(configuration, new VoidBlobStore()),
+ new HeartbeatServices(1000L, 1000L),
+
UnregisteredMetricGroups.createUnregisteredJobManagerMetricGroup(),
+ null,
+ new MemoryArchivedExecutionGraphStore(),
+ jobManagerRunnerFactory,
+ testingFatalErrorHandler,
+ fencingTokens);
+ }
- @VisibleForTesting
- CompletableFuture<Integer> getNumberJobs(Time timeout) {
- return callAsyncWithoutFencing(
- () -> listJobs(timeout).get().size(),
- timeout);
+ private static class HATestingDispatcher extends TestingDispatcher {
+
+ @Nullable
+ private final Queue<DispatcherId> fencingTokens;
+
+ HATestingDispatcher(
+ RpcService rpcService,
+ String endpointId,
+ Configuration configuration,
+ HighAvailabilityServices
highAvailabilityServices,
+ ResourceManagerGateway resourceManagerGateway,
+ BlobServer blobServer,
+ HeartbeatServices heartbeatServices,
+ JobManagerMetricGroup jobManagerMetricGroup,
+ @Nullable String metricQueryServicePath,
+ ArchivedExecutionGraphStore
archivedExecutionGraphStore,
+ JobManagerRunnerFactory jobManagerRunnerFactory,
+ FatalErrorHandler fatalErrorHandler,
+ @Nullable Queue<DispatcherId> fencingTokens)
throws Exception {
+ super(
+ rpcService,
+ endpointId,
+ configuration,
+ highAvailabilityServices,
+ resourceManagerGateway,
+ blobServer,
+ heartbeatServices,
+ jobManagerMetricGroup,
+ metricQueryServicePath,
+ archivedExecutionGraphStore,
+ jobManagerRunnerFactory,
+ fatalErrorHandler);
+ this.fencingTokens = fencingTokens;
}
@Override
protected void setFencingToken(@Nullable DispatcherId
newFencingToken) {
super.setFencingToken(newFencingToken);
+ final DispatcherId fencingToken;
if (newFencingToken == null) {
- fencingTokens.offer(NULL_FENCING_TOKEN);
+ fencingToken = NULL_FENCING_TOKEN;
} else {
- fencingTokens.offer(newFencingToken);
+ fencingToken = newFencingToken;
+ }
+
+ if (fencingTokens != null) {
Review comment:
Seems like we do not need nullable case anymore.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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
With regards,
Apache Git Services
