mynameborat commented on a change in pull request #1553:
URL: https://github.com/apache/samza/pull/1553#discussion_r751486018
##########
File path:
samza-core/src/main/java/org/apache/samza/coordinator/staticresource/StaticResourceJobCoordinator.java
##########
@@ -234,13 +262,33 @@ private void prepareWorkerExecution(JobModel jobModel,
JobCoordinatorMetadata ne
}
}
+ /**
+ * Wrapper around {@link MetadataResourceUtil} constructor so it can be
stubbed during testing.
+ */
@VisibleForTesting
MetadataResourceUtil metadataResourceUtil(JobModel jobModel) {
return new MetadataResourceUtil(jobModel, this.metrics, this.config);
}
+ /**
+ * Wrapper around {@link DiagnosticsUtil#buildDiagnosticsManager} so it can
be stubbed during testing.
+ */
+ @VisibleForTesting
+ Optional<DiagnosticsManager> buildDiagnosticsManager(String jobName,
+ String jobId, JobModel jobModel, String containerId, Optional<String>
execEnvContainerId, Config config) {
+ return DiagnosticsUtil.buildDiagnosticsManager(jobName, jobId, jobModel,
containerId, execEnvContainerId, config);
+ }
+
private Set<JobMetadataChange>
checkForMetadataChanges(JobCoordinatorMetadata newMetadata) {
JobCoordinatorMetadata previousMetadata =
this.jobCoordinatorMetadataManager.readJobCoordinatorMetadata();
return
this.jobCoordinatorMetadataManager.checkForMetadataChanges(newMetadata,
previousMetadata);
}
+
+ private static void quietlyStop(DiagnosticsManager diagnosticsManager) {
+ try {
+ diagnosticsManager.stop();
Review comment:
Also noticed that the `shutdownNow()` is never going to get invoked if
there is interruption since awaitTermination will throw interrupted exception
short circuiting the rest of the method.
--
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]
