This is an automated email from the ASF dual-hosted git repository.
cameronlee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/samza.git
The following commit(s) were added to refs/heads/master by this push:
new 3f38b21 SAMZA-2551: Upgrade all modules to automatically use
checkstyle 6.11.2 (part 5: includes samza-yarn, samza-rest) (#1396)
3f38b21 is described below
commit 3f38b217ce5a3bf11bc79703bc3733b420bd1089
Author: Cameron Lee <[email protected]>
AuthorDate: Mon Jun 29 16:15:51 2020 -0700
SAMZA-2551: Upgrade all modules to automatically use checkstyle 6.11.2
(part 5: includes samza-yarn, samza-rest) (#1396)
API/Upgrade/Usage changes: None
---
build.gradle | 12 +
.../java/org/apache/samza/monitor/JobsClient.java | 4 +-
.../samza/monitor/LocalStoreMonitorConfig.java | 2 +-
.../java/org/apache/samza/monitor/Monitor.java | 13 +-
.../org/apache/samza/monitor/MonitorLoader.java | 19 +-
.../apache/samza/monitor/SamzaMonitorService.java | 104 ++++-----
.../ScheduledExecutorSchedulingProvider.java | 20 +-
.../apache/samza/monitor/SchedulingProvider.java | 8 +-
.../org/apache/samza/rest/SamzaRestService.java | 2 +-
.../samza/rest/model/yarn/YarnApplicationInfo.java | 6 +-
.../samza/rest/proxy/job/AbstractJobProxy.java | 4 +-
.../org/apache/samza/rest/proxy/job/JobProxy.java | 7 +-
.../rest/proxy/job/YarnRestJobStatusProvider.java | 2 +-
.../samza/rest/proxy/task/SamzaTaskProxy.java | 10 +-
.../apache/samza/rest/proxy/task/TaskProxy.java | 2 -
.../samza/rest/proxy/task/TaskProxyFactory.java | 5 +-
.../samza/rest/proxy/task/TaskResourceConfig.java | 3 +-
.../samza/rest/resources/BaseResourceConfig.java | 4 +-
.../apache/samza/rest/resources/JobsResource.java | 5 +-
.../samza/rest/resources/JobsResourceConfig.java | 9 +-
.../apache/samza/rest/resources/TasksResource.java | 5 +-
.../samza/monitor/TestLocalStoreMonitor.java | 2 +-
.../apache/samza/monitor/TestMonitorService.java | 249 ++++++++++-----------
.../apache/samza/monitor/mock/DummyMonitor.java | 8 +-
.../monitor/mock/ExceptionThrowingMonitor.java | 8 +-
.../monitor/mock/InstantSchedulingProvider.java | 10 +-
.../samza/rest/resources/TestJobsResource.java | 8 +-
.../samza/rest/resources/TestTasksResource.java | 4 +-
.../rest/resources/mock/MockJobProxyFactory.java | 2 +-
.../rest/resources/mock/MockJobStatusProvider.java | 6 +-
.../samza/job/yarn/FileSystemImplConfig.java | 4 -
.../org/apache/samza/job/yarn/YarnAppState.java | 1 -
.../samza/job/yarn/YarnClusterResourceManager.java | 69 +++---
.../samza/validation/YarnJobValidationTool.java | 16 +-
.../samza/webapp/ApplicationMasterRestClient.java | 8 +-
.../org/apache/samza/config/TestYarnConfig.java | 2 -
.../samza/job/yarn/TestFileSystemImplConfig.java | 4 +-
.../job/yarn/TestLocalizerResourceConfig.java | 2 +-
.../job/yarn/TestLocalizerResourceMapper.java | 2 +-
.../job/yarn/TestYarnClusterResourceManager.java | 4 +-
.../apache/samza/job/yarn/TestYarnJobFactory.java | 6 +-
.../samza/job/yarn/util/MockContainerListener.java | 1 -
.../apache/samza/job/yarn/util/MockHttpServer.java | 2 +-
.../job/yarn/util/hadoop/TestHttpFileSystem.java | 8 +-
.../samza/validation/MockMetricsValidator.java | 8 +-
.../webapp/TestApplicationMasterRestClient.java | 2 +-
.../webapp/TestYarnContainerHeartbeatServlet.java | 18 +-
47 files changed, 348 insertions(+), 352 deletions(-)
diff --git a/build.gradle b/build.gradle
index 7268838..e8d6101 100644
--- a/build.gradle
+++ b/build.gradle
@@ -496,6 +496,7 @@ project(":samza-log4j2_$scalaSuffix") {
project(":samza-yarn_$scalaSuffix") {
apply plugin: 'scala'
apply plugin: 'lesscss'
+ apply plugin: 'checkstyle'
// Force scala joint compilation
sourceSets.main.scala.srcDir "src/main/java"
@@ -547,6 +548,11 @@ project(":samza-yarn_$scalaSuffix") {
testCompile "org.scalatest:scalatest_$scalaSuffix:$scalaTestVersion"
}
+ checkstyle {
+ configFile = new File(rootDir, "checkstyle/checkstyle.xml")
+ toolVersion = "$checkstyleVersion"
+ }
+
repositories {
maven {
url "http://repo.typesafe.com/typesafe/releases";
@@ -764,6 +770,7 @@ project(":samza-hdfs_$scalaSuffix") {
project(":samza-rest_$scalaSuffix") {
apply plugin: 'java'
+ apply plugin: 'checkstyle'
dependencies {
compile project(":samza-shell")
@@ -791,6 +798,11 @@ project(":samza-rest_$scalaSuffix") {
testCompile "org.mockito:mockito-core:$mockitoVersion"
}
+ checkstyle {
+ configFile = new File(rootDir, "checkstyle/checkstyle.xml")
+ toolVersion = "$checkstyleVersion"
+ }
+
tasks.create(name: "releaseRestServiceTar", type: Tar) {
description 'Build a tarball containing the samza-rest component and its
dependencies'
compression = Compression.GZIP
diff --git a/samza-rest/src/main/java/org/apache/samza/monitor/JobsClient.java
b/samza-rest/src/main/java/org/apache/samza/monitor/JobsClient.java
index 8d82600..4dc00b3 100644
--- a/samza-rest/src/main/java/org/apache/samza/monitor/JobsClient.java
+++ b/samza-rest/src/main/java/org/apache/samza/monitor/JobsClient.java
@@ -70,7 +70,7 @@ public class JobsClient {
*/
public List<Task> getTasks(JobInstance jobInstance) {
return queryJobStatusServers(baseUrl ->
String.format(ResourceConstants.GET_TASKS_URL, baseUrl,
- jobInstance.getJobName(), jobInstance.getJobId()), new
TypeReference<List<Task>>(){});
+ jobInstance.getJobName(), jobInstance.getJobId()), new
TypeReference<List<Task>>() { });
}
/**
@@ -81,7 +81,7 @@ public class JobsClient {
*/
public JobStatus getJobStatus(JobInstance jobInstance) {
Job job = queryJobStatusServers(baseUrl ->
String.format(ResourceConstants.GET_JOBS_URL, baseUrl,
- jobInstance.getJobName(), jobInstance.getJobId()), new
TypeReference<Job>(){});
+ jobInstance.getJobName(), jobInstance.getJobId()), new
TypeReference<Job>() { });
return job.getStatus();
}
diff --git
a/samza-rest/src/main/java/org/apache/samza/monitor/LocalStoreMonitorConfig.java
b/samza-rest/src/main/java/org/apache/samza/monitor/LocalStoreMonitorConfig.java
index f334c1f..6923f2f 100644
---
a/samza-rest/src/main/java/org/apache/samza/monitor/LocalStoreMonitorConfig.java
+++
b/samza-rest/src/main/java/org/apache/samza/monitor/LocalStoreMonitorConfig.java
@@ -88,7 +88,7 @@ public class LocalStoreMonitorConfig extends MapConfig {
* on the job status server.
*/
public List<String> getJobStatusServers() {
- return Arrays.asList(StringUtils.split(get(CONFIG_JOB_STATUS_SERVERS),
','));
+ return Arrays.asList(StringUtils.split(get(CONFIG_JOB_STATUS_SERVERS),
','));
}
/**
diff --git a/samza-rest/src/main/java/org/apache/samza/monitor/Monitor.java
b/samza-rest/src/main/java/org/apache/samza/monitor/Monitor.java
index 0d48213..4a05538 100644
--- a/samza-rest/src/main/java/org/apache/samza/monitor/Monitor.java
+++ b/samza-rest/src/main/java/org/apache/samza/monitor/Monitor.java
@@ -28,11 +28,10 @@ package org.apache.samza.monitor;
*/
public interface Monitor {
- /**
- * Do the work of the monitor. Because this can be arbitrary behavior up
to and including script execution,
- * IPC-related IOExceptions and concurrency-related InterruptedExceptions
are caught by the SamzaMonitorService.
- * @throws Exception if there was any problem running the monitor.
- */
- void monitor()
- throws Exception;
+ /**
+ * Do the work of the monitor. Because this can be arbitrary behavior up to
and including script execution,
+ * IPC-related IOExceptions and concurrency-related InterruptedExceptions
are caught by the SamzaMonitorService.
+ * @throws Exception if there was any problem running the monitor.
+ */
+ void monitor() throws Exception;
}
\ No newline at end of file
diff --git
a/samza-rest/src/main/java/org/apache/samza/monitor/MonitorLoader.java
b/samza-rest/src/main/java/org/apache/samza/monitor/MonitorLoader.java
index efbc29c..754ea9c 100644
--- a/samza-rest/src/main/java/org/apache/samza/monitor/MonitorLoader.java
+++ b/samza-rest/src/main/java/org/apache/samza/monitor/MonitorLoader.java
@@ -28,15 +28,14 @@ import org.apache.samza.util.ReflectionUtil;
public class MonitorLoader {
public static Monitor instantiateMonitor(String monitorName, MonitorConfig
monitorConfig,
- MetricsRegistry metricsRegistry)
- throws InstantiationException {
- String factoryClass = monitorConfig.getMonitorFactoryClass();
- try {
- MonitorFactory monitorFactory = ReflectionUtil.getObj(factoryClass,
MonitorFactory.class);
- return monitorFactory.getMonitorInstance(monitorName, monitorConfig,
metricsRegistry);
- } catch (Exception e) {
- throw (InstantiationException)
- new InstantiationException("Unable to instantiate monitor with
factory class " + factoryClass).initCause(e);
- }
+ MetricsRegistry metricsRegistry) throws InstantiationException {
+ String factoryClass = monitorConfig.getMonitorFactoryClass();
+ try {
+ MonitorFactory monitorFactory = ReflectionUtil.getObj(factoryClass,
MonitorFactory.class);
+ return monitorFactory.getMonitorInstance(monitorName, monitorConfig,
metricsRegistry);
+ } catch (Exception e) {
+ throw (InstantiationException)
+ new InstantiationException("Unable to instantiate monitor with
factory class " + factoryClass).initCause(e);
+ }
}
}
diff --git
a/samza-rest/src/main/java/org/apache/samza/monitor/SamzaMonitorService.java
b/samza-rest/src/main/java/org/apache/samza/monitor/SamzaMonitorService.java
index 05c4249..5558917 100644
--- a/samza-rest/src/main/java/org/apache/samza/monitor/SamzaMonitorService.java
+++ b/samza-rest/src/main/java/org/apache/samza/monitor/SamzaMonitorService.java
@@ -40,64 +40,64 @@ import static
org.apache.samza.monitor.MonitorLoader.instantiateMonitor;
*/
public class SamzaMonitorService {
- private static final Logger LOGGER =
LoggerFactory.getLogger(SamzaMonitorService.class);
- private static final SecureRandom RANDOM = new SecureRandom();
+ private static final Logger LOGGER =
LoggerFactory.getLogger(SamzaMonitorService.class);
+ private static final SecureRandom RANDOM = new SecureRandom();
- private final SchedulingProvider scheduler;
- private final SamzaRestConfig config;
- private final MetricsRegistry metricsRegistry;
+ private final SchedulingProvider scheduler;
+ private final SamzaRestConfig config;
+ private final MetricsRegistry metricsRegistry;
- public SamzaMonitorService(SamzaRestConfig config,
- MetricsRegistry metricsRegistry,
- SchedulingProvider schedulingProvider) {
- this.config = config;
- this.metricsRegistry = metricsRegistry;
- this.scheduler = schedulingProvider;
- }
+ public SamzaMonitorService(SamzaRestConfig config,
+ MetricsRegistry metricsRegistry,
+ SchedulingProvider schedulingProvider) {
+ this.config = config;
+ this.metricsRegistry = metricsRegistry;
+ this.scheduler = schedulingProvider;
+ }
- public void start() {
- try {
- Map<String, MonitorConfig> monitorConfigs =
getMonitorConfigs(config);
- for (Map.Entry<String, MonitorConfig> entry :
monitorConfigs.entrySet()) {
- String monitorName = entry.getKey();
- MonitorConfig monitorConfig = entry.getValue();
+ public void start() {
+ try {
+ Map<String, MonitorConfig> monitorConfigs = getMonitorConfigs(config);
+ for (Map.Entry<String, MonitorConfig> entry : monitorConfigs.entrySet())
{
+ String monitorName = entry.getKey();
+ MonitorConfig monitorConfig = entry.getValue();
- if
(!Strings.isNullOrEmpty(monitorConfig.getMonitorFactoryClass())) {
- int schedulingIntervalInMs =
monitorConfig.getSchedulingIntervalInMs();
- int monitorSchedulingJitterInMs = (int)
(RANDOM.nextInt(schedulingIntervalInMs + 1) *
(monitorConfig.getSchedulingJitterPercent() / 100.0));
- schedulingIntervalInMs += monitorSchedulingJitterInMs;
- LOGGER.info("Scheduling the monitor: {} to run every {}
ms.", monitorName, schedulingIntervalInMs);
-
scheduler.schedule(getRunnable(instantiateMonitor(monitorName, monitorConfig,
metricsRegistry)),
- schedulingIntervalInMs);
- } else {
- // When MonitorFactoryClass is not defined in the config,
ignore the monitor config
- LOGGER.warn("Not scheduling the monitor: {} to run, since
monitor factory class is not set in config.", monitorName);
- }
- }
- } catch (InstantiationException e) {
- LOGGER.error("Exception when instantiating the monitor : ", e);
- throw new SamzaException(e);
+ if (!Strings.isNullOrEmpty(monitorConfig.getMonitorFactoryClass())) {
+ int schedulingIntervalInMs =
monitorConfig.getSchedulingIntervalInMs();
+ int monitorSchedulingJitterInMs = (int)
(RANDOM.nextInt(schedulingIntervalInMs + 1) *
(monitorConfig.getSchedulingJitterPercent() / 100.0));
+ schedulingIntervalInMs += monitorSchedulingJitterInMs;
+ LOGGER.info("Scheduling the monitor: {} to run every {} ms.",
monitorName, schedulingIntervalInMs);
+ scheduler.schedule(getRunnable(instantiateMonitor(monitorName,
monitorConfig, metricsRegistry)),
+ schedulingIntervalInMs);
+ } else {
+ // When MonitorFactoryClass is not defined in the config, ignore the
monitor config
+ LOGGER.warn("Not scheduling the monitor: {} to run, since monitor
factory class is not set in config.", monitorName);
}
+ }
+ } catch (InstantiationException e) {
+ LOGGER.error("Exception when instantiating the monitor : ", e);
+ throw new SamzaException(e);
}
+ }
- public void stop() {
- this.scheduler.stop();
- }
+ public void stop() {
+ this.scheduler.stop();
+ }
- private Runnable getRunnable(final Monitor monitor) {
- return new Runnable() {
- public void run() {
- try {
- monitor.monitor();
- } catch (IOException e) {
- LOGGER.error("Caught IOException during " +
monitor.toString() + ".monitor()", e);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- LOGGER.error("Caught InterruptedException during " +
monitor.toString() + ".monitor()", e);
- } catch (Exception e) {
- LOGGER.error("Unexpected exception during {}.monitor()",
monitor, e);
- }
- }
- };
- }
+ private Runnable getRunnable(final Monitor monitor) {
+ return new Runnable() {
+ public void run() {
+ try {
+ monitor.monitor();
+ } catch (IOException e) {
+ LOGGER.error("Caught IOException during " + monitor.toString() +
".monitor()", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOGGER.error("Caught InterruptedException during " +
monitor.toString() + ".monitor()", e);
+ } catch (Exception e) {
+ LOGGER.error("Unexpected exception during {}.monitor()", monitor, e);
+ }
+ }
+ };
+ }
}
diff --git
a/samza-rest/src/main/java/org/apache/samza/monitor/ScheduledExecutorSchedulingProvider.java
b/samza-rest/src/main/java/org/apache/samza/monitor/ScheduledExecutorSchedulingProvider.java
index c0c448c..e1f0711 100644
---
a/samza-rest/src/main/java/org/apache/samza/monitor/ScheduledExecutorSchedulingProvider.java
+++
b/samza-rest/src/main/java/org/apache/samza/monitor/ScheduledExecutorSchedulingProvider.java
@@ -24,17 +24,17 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
public class ScheduledExecutorSchedulingProvider implements SchedulingProvider
{
- private final ScheduledExecutorService scheduler;
+ private final ScheduledExecutorService scheduler;
- public ScheduledExecutorSchedulingProvider(ScheduledExecutorService
scheduler) {
- this.scheduler = scheduler;
- }
+ public ScheduledExecutorSchedulingProvider(ScheduledExecutorService
scheduler) {
+ this.scheduler = scheduler;
+ }
- public void schedule(Runnable runnable, int interval) {
- this.scheduler.scheduleAtFixedRate(runnable, 0, interval,
MILLISECONDS);
- }
+ public void schedule(Runnable runnable, int interval) {
+ this.scheduler.scheduleAtFixedRate(runnable, 0, interval, MILLISECONDS);
+ }
- public void stop() {
- this.scheduler.shutdownNow();
- }
+ public void stop() {
+ this.scheduler.shutdownNow();
+ }
}
diff --git
a/samza-rest/src/main/java/org/apache/samza/monitor/SchedulingProvider.java
b/samza-rest/src/main/java/org/apache/samza/monitor/SchedulingProvider.java
index 627a333..d68f6c1 100644
--- a/samza-rest/src/main/java/org/apache/samza/monitor/SchedulingProvider.java
+++ b/samza-rest/src/main/java/org/apache/samza/monitor/SchedulingProvider.java
@@ -22,9 +22,9 @@ package org.apache.samza.monitor;
* Provides scheduling functionality to the SamzaMonitorService.
*/
public interface SchedulingProvider {
- /* Schedule a the given Runnable to run() every INTERVAL ms. */
- void schedule(Runnable runnable, int intervalMs);
+ /* Schedule a the given Runnable to run() every INTERVAL ms. */
+ void schedule(Runnable runnable, int intervalMs);
- /* Stop any future executions of any scheduled tasks. */
- void stop();
+ /* Stop any future executions of any scheduled tasks. */
+ void stop();
}
diff --git
a/samza-rest/src/main/java/org/apache/samza/rest/SamzaRestService.java
b/samza-rest/src/main/java/org/apache/samza/rest/SamzaRestService.java
index b5a75d7..c5d6af7 100644
--- a/samza-rest/src/main/java/org/apache/samza/rest/SamzaRestService.java
+++ b/samza-rest/src/main/java/org/apache/samza/rest/SamzaRestService.java
@@ -120,7 +120,7 @@ public class SamzaRestService {
} catch (Throwable t) {
log.error("Exception in main.", t);
} finally {
- if (schedulingProvider != null){
+ if (schedulingProvider != null) {
schedulingProvider.stop();
}
}
diff --git
a/samza-rest/src/main/java/org/apache/samza/rest/model/yarn/YarnApplicationInfo.java
b/samza-rest/src/main/java/org/apache/samza/rest/model/yarn/YarnApplicationInfo.java
index 8d55c89..a696999 100644
---
a/samza-rest/src/main/java/org/apache/samza/rest/model/yarn/YarnApplicationInfo.java
+++
b/samza-rest/src/main/java/org/apache/samza/rest/model/yarn/YarnApplicationInfo.java
@@ -19,8 +19,6 @@
package org.apache.samza.rest.model.yarn;
import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
import org.apache.samza.rest.proxy.job.JobInstance;
import org.codehaus.jackson.annotate.JsonIgnoreProperties;
import org.codehaus.jackson.annotate.JsonProperty;
@@ -56,8 +54,8 @@ public class YarnApplicationInfo {
* @return the job name to use for the job in YARN.
*/
public static String getQualifiedJobName(JobInstance jobInstance) {
- final String JOB_NAME_ID_FORMAT = "%s_%s";
- return String.format(JOB_NAME_ID_FORMAT, jobInstance.getJobName(),
jobInstance.getJobId());
+ final String jobNameIdFormat = "%s_%s";
+ return String.format(jobNameIdFormat, jobInstance.getJobName(),
jobInstance.getJobId());
}
@JsonIgnoreProperties(ignoreUnknown = true)
diff --git
a/samza-rest/src/main/java/org/apache/samza/rest/proxy/job/AbstractJobProxy.java
b/samza-rest/src/main/java/org/apache/samza/rest/proxy/job/AbstractJobProxy.java
index 8b35352..f409374 100644
---
a/samza-rest/src/main/java/org/apache/samza/rest/proxy/job/AbstractJobProxy.java
+++
b/samza-rest/src/main/java/org/apache/samza/rest/proxy/job/AbstractJobProxy.java
@@ -74,8 +74,8 @@ public abstract class AbstractJobProxy implements JobProxy {
throws IOException, InterruptedException {
List<Job> allJobs = new ArrayList<>();
Collection<JobInstance> jobInstances = getAllJobInstances();
- for(JobInstance jobInstance : jobInstances) {
- allJobs.add(new Job(jobInstance.getJobName(), jobInstance.getJobId()));
+ for (JobInstance jobInstance : jobInstances) {
+ allJobs.add(new Job(jobInstance.getJobName(), jobInstance.getJobId()));
}
getJobStatusProvider().getJobStatuses(allJobs);
diff --git
a/samza-rest/src/main/java/org/apache/samza/rest/proxy/job/JobProxy.java
b/samza-rest/src/main/java/org/apache/samza/rest/proxy/job/JobProxy.java
index 7e168d7..d42ff7a 100644
--- a/samza-rest/src/main/java/org/apache/samza/rest/proxy/job/JobProxy.java
+++ b/samza-rest/src/main/java/org/apache/samza/rest/proxy/job/JobProxy.java
@@ -21,7 +21,6 @@ package org.apache.samza.rest.proxy.job;
import java.io.IOException;
import java.util.List;
import org.apache.samza.rest.model.Job;
-import org.apache.samza.rest.model.JobStatus;
/**
@@ -57,8 +56,8 @@ public interface JobProxy {
/**
* Starts the job instance specified by jobName and jobId. When this method
returns, the status of the job
- * should be {@link JobStatus#STARTING} or
- * {@link JobStatus#STARTED} depending on the implementation.
+ * should be {@link org.apache.samza.rest.model.JobStatus#STARTING} or
+ * {@link org.apache.samza.rest.model.JobStatus#STARTED} depending on the
implementation.
*
* @param jobInstance the instance of the job to start.
* @throws Exception if the job could not be successfully started.
@@ -68,7 +67,7 @@ public interface JobProxy {
/**
* Stops the job instance specified by jobName and jobId. When this method
returns, the status of the job
- * should be {@link JobStatus#STOPPED}.
+ * should be {@link org.apache.samza.rest.model.JobStatus#STOPPED}.
*
* @param jobInstance the instance of the job to stop.
* @throws Exception if the job could not be successfully stopped.
diff --git
a/samza-rest/src/main/java/org/apache/samza/rest/proxy/job/YarnRestJobStatusProvider.java
b/samza-rest/src/main/java/org/apache/samza/rest/proxy/job/YarnRestJobStatusProvider.java
index 1009f10..e0b79ef 100644
---
a/samza-rest/src/main/java/org/apache/samza/rest/proxy/job/YarnRestJobStatusProvider.java
+++
b/samza-rest/src/main/java/org/apache/samza/rest/proxy/job/YarnRestJobStatusProvider.java
@@ -70,7 +70,7 @@ public class YarnRestJobStatusProvider implements
JobStatusProvider {
// We will identify the YARN application states by their qualified names,
so build a map
// to translate back from that name to the JobInfo we wish to populate.
final Map<String, Job> qualifiedJobToInfo = new HashMap<>();
- for(Job job : jobs) {
+ for (Job job : jobs) {
qualifiedJobToInfo.put(YarnApplicationInfo.getQualifiedJobName(new
JobInstance(job.getJobName(), job.getJobId())), job);
}
diff --git
a/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/SamzaTaskProxy.java
b/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/SamzaTaskProxy.java
index 31931db..b86da1f 100644
---
a/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/SamzaTaskProxy.java
+++
b/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/SamzaTaskProxy.java
@@ -143,10 +143,10 @@ public class SamzaTaskProxy implements TaskProxy {
StorageConfig storageConfig = new StorageConfig(consumer.getConfig());
List<String> storeNames = storageConfig.getStoreNames();
return taskNameToContainerIdMapping.entrySet()
- .stream()
- .map(entry -> {
- String hostName =
containerIdToHostMapping.get(entry.getValue()).get(SetContainerHostMapping.HOST_KEY);
- return new Task(hostName, entry.getKey(), entry.getValue(), new
ArrayList<>(), storeNames);
- }).collect(Collectors.toList());
+ .stream()
+ .map(entry -> {
+ String hostName =
containerIdToHostMapping.get(entry.getValue()).get(SetContainerHostMapping.HOST_KEY);
+ return new Task(hostName, entry.getKey(), entry.getValue(), new
ArrayList<>(), storeNames);
+ }).collect(Collectors.toList());
}
}
diff --git
a/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/TaskProxy.java
b/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/TaskProxy.java
index 54da8c9..46bf385 100644
--- a/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/TaskProxy.java
+++ b/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/TaskProxy.java
@@ -22,8 +22,6 @@ import java.io.IOException;
import java.util.List;
import org.apache.samza.rest.model.Task;
import org.apache.samza.rest.proxy.job.JobInstance;
-import org.apache.samza.rest.proxy.job.JobProxy;
-import org.apache.samza.rest.resources.JobsResourceConfig;
/**
diff --git
a/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/TaskProxyFactory.java
b/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/TaskProxyFactory.java
index 5e9957f..639e983 100644
---
a/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/TaskProxyFactory.java
+++
b/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/TaskProxyFactory.java
@@ -18,8 +18,6 @@
*/
package org.apache.samza.rest.proxy.task;
-import org.apache.samza.config.Config;
-
/**
* Factory interface that will be used to create {@link TaskProxy}
@@ -33,8 +31,7 @@ import org.apache.samza.config.Config;
public interface TaskProxyFactory {
/**
- *
- * @param config the {@link Config} to pass to the proxy.
+ * @param config the {@link TaskResourceConfig} to pass to the proxy.
* @return the created proxy.
*/
TaskProxy getTaskProxy(TaskResourceConfig config);
diff --git
a/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/TaskResourceConfig.java
b/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/TaskResourceConfig.java
index 40cf706..401d4ac 100644
---
a/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/TaskResourceConfig.java
+++
b/samza-rest/src/main/java/org/apache/samza/rest/proxy/task/TaskResourceConfig.java
@@ -20,11 +20,10 @@ package org.apache.samza.rest.proxy.task;
import org.apache.samza.config.Config;
import org.apache.samza.rest.resources.BaseResourceConfig;
-import org.apache.samza.rest.resources.TasksResource;
/**
- * Configurations for the {@link TasksResource} endpoint.
+ * Configurations for the {@link
org.apache.samza.rest.resources.TasksResource} endpoint.
*/
public class TaskResourceConfig extends BaseResourceConfig {
diff --git
a/samza-rest/src/main/java/org/apache/samza/rest/resources/BaseResourceConfig.java
b/samza-rest/src/main/java/org/apache/samza/rest/resources/BaseResourceConfig.java
index eca8fdc..1c4d95a 100644
---
a/samza-rest/src/main/java/org/apache/samza/rest/resources/BaseResourceConfig.java
+++
b/samza-rest/src/main/java/org/apache/samza/rest/resources/BaseResourceConfig.java
@@ -21,7 +21,6 @@ package org.apache.samza.rest.resources;
import org.apache.samza.config.Config;
import org.apache.samza.config.MapConfig;
import org.apache.samza.config.factories.PropertiesConfigFactory;
-import org.apache.samza.rest.proxy.installation.InstallationRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,7 +40,8 @@ public class BaseResourceConfig extends MapConfig {
/**
* The path where all the Samza jobs are installed (unzipped). Each
subdirectory of this path
- * is expected to be a Samza job installation and corresponds to one {@link
InstallationRecord}.
+ * is expected to be a Samza job installation and corresponds to one
+ * {@link org.apache.samza.rest.proxy.installation.InstallationRecord}.
*/
public static final String CONFIG_JOB_INSTALLATIONS_PATH =
"job.installations.path";
diff --git
a/samza-rest/src/main/java/org/apache/samza/rest/resources/JobsResource.java
b/samza-rest/src/main/java/org/apache/samza/rest/resources/JobsResource.java
index caad56c..d79e2fa 100644
--- a/samza-rest/src/main/java/org/apache/samza/rest/resources/JobsResource.java
+++ b/samza-rest/src/main/java/org/apache/samza/rest/resources/JobsResource.java
@@ -33,7 +33,6 @@ import org.apache.samza.rest.model.JobStatus;
import org.apache.samza.rest.proxy.job.AbstractJobProxy;
import org.apache.samza.rest.proxy.job.JobInstance;
import org.apache.samza.rest.proxy.job.JobProxy;
-import org.apache.samza.rest.proxy.job.JobProxyFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,9 +50,9 @@ public class JobsResource {
/**
* Initializes a JobResource with {@link JobProxy} from the
- * {@link JobProxyFactory} class specified in the configuration.
+ * {@link org.apache.samza.rest.proxy.job.JobProxyFactory} class specified
in the configuration.
*
- * @param config the configuration containing the {@link JobProxyFactory}
class.
+ * @param config the configuration containing the {@link
org.apache.samza.rest.proxy.job.JobProxyFactory} class.
*/
public JobsResource(JobsResourceConfig config) {
jobProxy = AbstractJobProxy.fromFactory(config);
diff --git
a/samza-rest/src/main/java/org/apache/samza/rest/resources/JobsResourceConfig.java
b/samza-rest/src/main/java/org/apache/samza/rest/resources/JobsResourceConfig.java
index bd52e65..7748fee 100644
---
a/samza-rest/src/main/java/org/apache/samza/rest/resources/JobsResourceConfig.java
+++
b/samza-rest/src/main/java/org/apache/samza/rest/resources/JobsResourceConfig.java
@@ -19,8 +19,6 @@
package org.apache.samza.rest.resources;
import org.apache.samza.config.Config;
-import org.apache.samza.rest.proxy.job.JobProxy;
-import org.apache.samza.rest.proxy.job.JobProxyFactory;
/**
@@ -29,8 +27,8 @@ import org.apache.samza.rest.proxy.job.JobProxyFactory;
public class JobsResourceConfig extends BaseResourceConfig {
/**
- * Specifies the canonical name of the {@link JobProxyFactory} class to
produce
- * {@link JobProxy} instances.
+ * Specifies the canonical name of the {@link
org.apache.samza.rest.proxy.job.JobProxyFactory} class to produce
+ * {@link org.apache.samza.rest.proxy.job.JobProxy} instances.
*
* To use your own proxy, implement the factory and specify the class for
this config.
*/
@@ -42,7 +40,8 @@ public class JobsResourceConfig extends BaseResourceConfig {
/**
* @see JobsResourceConfig#CONFIG_JOB_PROXY_FACTORY
- * @return the canonical name of the {@link JobProxyFactory} class to
produce {@link JobProxy} instances.
+ * @return the canonical name of the {@link
org.apache.samza.rest.proxy.job.JobProxyFactory} class to produce
+ * {@link org.apache.samza.rest.proxy.job.JobProxy} instances.
*/
public String getJobProxyFactory() {
return get(CONFIG_JOB_PROXY_FACTORY);
diff --git
a/samza-rest/src/main/java/org/apache/samza/rest/resources/TasksResource.java
b/samza-rest/src/main/java/org/apache/samza/rest/resources/TasksResource.java
index ff6cc26..12db787 100644
---
a/samza-rest/src/main/java/org/apache/samza/rest/resources/TasksResource.java
+++
b/samza-rest/src/main/java/org/apache/samza/rest/resources/TasksResource.java
@@ -28,7 +28,6 @@ import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import org.apache.commons.lang.StringUtils;
import org.apache.samza.SamzaException;
-import org.apache.samza.rest.model.Task;
import org.apache.samza.rest.proxy.job.JobInstance;
import org.apache.samza.rest.proxy.task.TaskProxyFactory;
import org.apache.samza.rest.proxy.task.TaskProxy;
@@ -68,11 +67,11 @@ public class TasksResource {
}
/**
- * Gets the list of {@link Task} for the job instance specified by jobName
and jobId.
+ * Gets the list of {@link org.apache.samza.rest.model.Task} for the job
instance specified by jobName and jobId.
* @param jobName the name of the job as configured in {@link
org.apache.samza.config.JobConfig#JOB_NAME}
* @param jobId the id of the job as configured in {@link
org.apache.samza.config.JobConfig#JOB_ID}.
* @return a {@link javax.ws.rs.core.Response.Status#OK} {@link
javax.ws.rs.core.Response}
- * contains a list of {@link Task}, where each task belongs to
+ * contains a list of {@link org.apache.samza.rest.model.Task},
where each task belongs to
* the samza job. {@link
javax.ws.rs.core.Response.Status#BAD_REQUEST} is returned for invalid
* job instances.
*/
diff --git
a/samza-rest/src/test/java/org/apache/samza/monitor/TestLocalStoreMonitor.java
b/samza-rest/src/test/java/org/apache/samza/monitor/TestLocalStoreMonitor.java
index 15f0df9..972598f 100644
---
a/samza-rest/src/test/java/org/apache/samza/monitor/TestLocalStoreMonitor.java
+++
b/samza-rest/src/test/java/org/apache/samza/monitor/TestLocalStoreMonitor.java
@@ -46,7 +46,7 @@ import static junit.framework.TestCase.*;
public class TestLocalStoreMonitor {
- private static Logger LOG =
LoggerFactory.getLogger(TestLocalStoreMonitor.class);
+ private static final Logger LOG =
LoggerFactory.getLogger(TestLocalStoreMonitor.class);
private File localStoreDir;
diff --git
a/samza-rest/src/test/java/org/apache/samza/monitor/TestMonitorService.java
b/samza-rest/src/test/java/org/apache/samza/monitor/TestMonitorService.java
index c7a991b..b2924fc 100644
--- a/samza-rest/src/test/java/org/apache/samza/monitor/TestMonitorService.java
+++ b/samza-rest/src/test/java/org/apache/samza/monitor/TestMonitorService.java
@@ -46,137 +46,136 @@ import static org.junit.Assert.fail;
public class TestMonitorService {
- private static final MetricsRegistry METRICS_REGISTRY = new
NoOpMetricsRegistry();
-
- @Test
- public void testMonitorsShouldBeInstantiatedProperly() {
- // Test that a monitor should be instantiated properly by invoking
- // the appropriate factory method.
- Map<String, String> configMap =
ImmutableMap.of(CONFIG_MONITOR_FACTORY_CLASS,
-
DummyMonitorFactory.class.getCanonicalName());
- Monitor monitor = null;
- try {
- monitor = MonitorLoader.instantiateMonitor("testMonitor", new
MonitorConfig(new MapConfig(configMap)),
- METRICS_REGISTRY);
- } catch (InstantiationException e) {
- fail();
- }
- assertNotNull(monitor);
- // Object should implement the monitor().
- try {
- monitor.monitor();
- } catch (Exception e) {
- fail();
- }
+ private static final MetricsRegistry METRICS_REGISTRY = new
NoOpMetricsRegistry();
+
+ @Test
+ public void testMonitorsShouldBeInstantiatedProperly() {
+ // Test that a monitor should be instantiated properly by invoking
+ // the appropriate factory method.
+ Map<String, String> configMap =
ImmutableMap.of(CONFIG_MONITOR_FACTORY_CLASS,
+
DummyMonitorFactory.class.getCanonicalName());
+ Monitor monitor = null;
+ try {
+ monitor = MonitorLoader.instantiateMonitor("testMonitor", new
MonitorConfig(new MapConfig(configMap)),
+ METRICS_REGISTRY);
+ } catch (InstantiationException e) {
+ fail();
}
-
- @Test
- public void testShouldGroupRelevantMonitorConfigTogether() {
- // Test that Monitor Loader groups relevant config together.
- Map<String, String> firstMonitorConfig =
ImmutableMap.of("monitor.monitor1.factory.class",
-
"org.apache.samza.monitor.DummyMonitor",
-
"monitor.monitor1.scheduling.interval.ms",
- "100");
- Map<String, String> secondMonitorConfig =
ImmutableMap.of("monitor.monitor2.factory.class",
-
"org.apache.samza.monitor.DummyMonitor",
-
"monitor.monitor2.scheduling.interval.ms",
- "200");
- MapConfig mapConfig = new
MapConfig(ImmutableList.of(firstMonitorConfig, secondMonitorConfig));
- MonitorConfig expectedFirstConfig = new MonitorConfig(new
MapConfig(firstMonitorConfig).subset("monitor.monitor1."));
- MonitorConfig expectedSecondConfig = new MonitorConfig(new
MapConfig(secondMonitorConfig).subset("monitor.monitor2."));
- Map<String, MonitorConfig> expected = ImmutableMap.of("monitor1",
expectedFirstConfig, "monitor2", expectedSecondConfig);
- assertEquals(expected, MonitorConfig.getMonitorConfigs(mapConfig));
+ assertNotNull(monitor);
+ // Object should implement the monitor().
+ try {
+ monitor.monitor();
+ } catch (Exception e) {
+ fail();
}
-
- @Test
- public void testMonitorExceptionIsolation() {
- // Test that an exception from a monitor doesn't bubble up out of the
scheduler.
- Map<String, String> configMap =
- ImmutableMap.of(String.format("monitor.name.%s",
CONFIG_MONITOR_FACTORY_CLASS),
-
ExceptionThrowingMonitorFactory.class.getCanonicalName());
- SamzaRestConfig config = new SamzaRestConfig(new MapConfig(configMap));
- SamzaMonitorService monitorService = new SamzaMonitorService(config,
-
METRICS_REGISTRY,
- new
InstantSchedulingProvider());
-
- // This will throw if the exception isn't caught within the provider.
- monitorService.start();
- monitorService.stop();
+ }
+
+ @Test
+ public void testShouldGroupRelevantMonitorConfigTogether() {
+ // Test that Monitor Loader groups relevant config together.
+ Map<String, String> firstMonitorConfig =
ImmutableMap.of("monitor.monitor1.factory.class",
+
"org.apache.samza.monitor.DummyMonitor",
+
"monitor.monitor1.scheduling.interval.ms",
+ "100");
+ Map<String, String> secondMonitorConfig =
ImmutableMap.of("monitor.monitor2.factory.class",
+
"org.apache.samza.monitor.DummyMonitor",
+
"monitor.monitor2.scheduling.interval.ms",
+ "200");
+ MapConfig mapConfig = new MapConfig(ImmutableList.of(firstMonitorConfig,
secondMonitorConfig));
+ MonitorConfig expectedFirstConfig = new MonitorConfig(new
MapConfig(firstMonitorConfig).subset("monitor.monitor1."));
+ MonitorConfig expectedSecondConfig = new MonitorConfig(new
MapConfig(secondMonitorConfig).subset("monitor.monitor2."));
+ Map<String, MonitorConfig> expected = ImmutableMap.of("monitor1",
expectedFirstConfig, "monitor2", expectedSecondConfig);
+ assertEquals(expected, MonitorConfig.getMonitorConfigs(mapConfig));
+ }
+
+ @Test
+ public void testMonitorExceptionIsolation() {
+ // Test that an exception from a monitor doesn't bubble up out of the
scheduler.
+ Map<String, String> configMap =
+ ImmutableMap.of(String.format("monitor.name.%s",
CONFIG_MONITOR_FACTORY_CLASS),
+
ExceptionThrowingMonitorFactory.class.getCanonicalName());
+ SamzaRestConfig config = new SamzaRestConfig(new MapConfig(configMap));
+ SamzaMonitorService monitorService = new SamzaMonitorService(config,
+
METRICS_REGISTRY,
+ new
InstantSchedulingProvider());
+
+ // This will throw if the exception isn't caught within the provider.
+ monitorService.start();
+ monitorService.stop();
+ }
+
+ @Test
+ public void testShouldNotFailWhenTheMonitorFactoryClassIsNotDefined()
+ throws Exception {
+ // Test that when MonitorFactoryClass is not defined in the config,
monitor service
+ // should not fail.
+ Map<String, String> configMap =
ImmutableMap.of("monitor.monitor1.config.key1", "configValue1",
+
"monitor.monitor1.config.key2", "configValue2",
+
String.format("monitor.MOCK_MONITOR.%s", CONFIG_MONITOR_FACTORY_CLASS),
+
MockMonitorFactory.class.getCanonicalName());
+
+ SamzaRestConfig config = new SamzaRestConfig(new MapConfig(configMap));
+ SamzaMonitorService monitorService = new SamzaMonitorService(config,
+
METRICS_REGISTRY,
+ new
InstantSchedulingProvider());
+ try {
+ monitorService.start();
+ } catch (Exception e) {
+ fail();
}
-
- @Test
- public void testShouldNotFailWhenTheMonitorFactoryClassIsNotDefined()
- throws Exception {
- // Test that when MonitorFactoryClass is not defined in the config,
monitor service
- // should not fail.
- Map<String, String> configMap =
ImmutableMap.of("monitor.monitor1.config.key1", "configValue1",
-
"monitor.monitor1.config.key2", "configValue2",
-
String.format("monitor.MOCK_MONITOR.%s", CONFIG_MONITOR_FACTORY_CLASS),
-
MockMonitorFactory.class.getCanonicalName());
-
- SamzaRestConfig config = new SamzaRestConfig(new MapConfig(configMap));
- SamzaMonitorService monitorService = new SamzaMonitorService(config,
-
METRICS_REGISTRY,
- new
InstantSchedulingProvider());
+ Mockito.verify(MockMonitorFactory.MOCK_MONITOR,
Mockito.times(1)).monitor();
+ }
+
+ @Test(expected = SamzaException.class)
+ public void testShouldFailWhenTheMonitorFactoryClassIsInvalid() {
+ // Test that when MonitorFactoryClass is defined in the config and is
invalid,
+ // monitor service should fail. Should throw back SamzaException.
+ Map<String, String> configMap =
ImmutableMap.of(String.format("monitor.name.%s", CONFIG_MONITOR_FACTORY_CLASS),
+ "RandomClassName");
+ SamzaRestConfig config = new SamzaRestConfig(new MapConfig(configMap));
+ SamzaMonitorService monitorService = new SamzaMonitorService(config,
+
METRICS_REGISTRY,
+ new
InstantSchedulingProvider());
+ monitorService.start();
+ }
+
+ @Test
+ public void testScheduledExecutorSchedulingProvider() {
+ // Test that the monitor is scheduled by the
ScheduledExecutorSchedulingProvider
+ ScheduledExecutorService executorService =
Executors.newScheduledThreadPool(1);
+ ScheduledExecutorSchedulingProvider provider = new
ScheduledExecutorSchedulingProvider(executorService);
+
+ // notifyingMonitor.monitor() should be called repeatedly.
+ final CountDownLatch wasCalledLatch = new CountDownLatch(3);
+
+ final Monitor notifyingMonitor = new Monitor() {
+ @Override
+ public void monitor() {
+ wasCalledLatch.countDown();
+ }
+ };
+
+ Runnable runnableMonitor = new Runnable() {
+ public void run() {
try {
- monitorService.start();
+ notifyingMonitor.monitor();
} catch (Exception e) {
- fail();
+ // Must be caught because they are checked in monitor()
+ fail();
}
- Mockito.verify(MockMonitorFactory.MOCK_MONITOR,
Mockito.times(1)).monitor();
+ }
+ };
+
+ // monitor should get called every 1ms, so if await() misses the first
call, there will be more.
+ provider.schedule(runnableMonitor, 1);
+
+ try {
+ assertTrue(wasCalledLatch.await(5L, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ executorService.shutdownNow();
}
- @Test(expected = SamzaException.class)
- public void testShouldFailWhenTheMonitorFactoryClassIsInvalid() {
- // Test that when MonitorFactoryClass is defined in the config and is
invalid,
- // monitor service should fail. Should throw back SamzaException.
- Map<String, String> configMap =
ImmutableMap.of(String.format("monitor.name.%s", CONFIG_MONITOR_FACTORY_CLASS),
- "RandomClassName");
- SamzaRestConfig config = new SamzaRestConfig(new MapConfig(configMap));
- SamzaMonitorService monitorService = new SamzaMonitorService(config,
-
METRICS_REGISTRY,
- new
InstantSchedulingProvider());
- monitorService.start();
- }
-
- @Test
- public void testScheduledExecutorSchedulingProvider() {
- // Test that the monitor is scheduled by the
ScheduledExecutorSchedulingProvider
- ScheduledExecutorService executorService =
Executors.newScheduledThreadPool(1);
- ScheduledExecutorSchedulingProvider provider =
- new ScheduledExecutorSchedulingProvider(executorService);
-
- // notifyingMonitor.monitor() should be called repeatedly.
- final CountDownLatch wasCalledLatch = new CountDownLatch(3);
-
- final Monitor notifyingMonitor = new Monitor() {
- @Override
- public void monitor() {
- wasCalledLatch.countDown();
- }
- };
-
- Runnable runnableMonitor = new Runnable() {
- public void run() {
- try {
- notifyingMonitor.monitor();
- } catch (Exception e) {
- // Must be caught because they are checked in monitor()
- fail();
- }
- }
- };
-
- // monitor should get called every 1ms, so if await() misses the first
call, there will be more.
- provider.schedule(runnableMonitor, 1);
-
- try {
- assertTrue(wasCalledLatch.await(5l, TimeUnit.SECONDS));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- } finally {
- executorService.shutdownNow();
- }
-
- }
+ }
}
\ No newline at end of file
diff --git
a/samza-rest/src/test/java/org/apache/samza/monitor/mock/DummyMonitor.java
b/samza-rest/src/test/java/org/apache/samza/monitor/mock/DummyMonitor.java
index c6a2b28..1b28289 100644
--- a/samza-rest/src/test/java/org/apache/samza/monitor/mock/DummyMonitor.java
+++ b/samza-rest/src/test/java/org/apache/samza/monitor/mock/DummyMonitor.java
@@ -22,8 +22,8 @@ import org.apache.samza.monitor.Monitor;
public class DummyMonitor implements Monitor {
- @Override
- public void monitor() {
- // Do nothing!
- }
+ @Override
+ public void monitor() {
+ // Do nothing!
+ }
}
diff --git
a/samza-rest/src/test/java/org/apache/samza/monitor/mock/ExceptionThrowingMonitor.java
b/samza-rest/src/test/java/org/apache/samza/monitor/mock/ExceptionThrowingMonitor.java
index 035e2ed..aa1bcda 100644
---
a/samza-rest/src/test/java/org/apache/samza/monitor/mock/ExceptionThrowingMonitor.java
+++
b/samza-rest/src/test/java/org/apache/samza/monitor/mock/ExceptionThrowingMonitor.java
@@ -24,8 +24,8 @@ import java.io.IOException;
public class ExceptionThrowingMonitor implements Monitor {
- @Override
- public void monitor() throws IOException {
- throw new IOException("I don't know what I was expecting.");
- }
+ @Override
+ public void monitor() throws IOException {
+ throw new IOException("I don't know what I was expecting.");
+ }
}
diff --git
a/samza-rest/src/test/java/org/apache/samza/monitor/mock/InstantSchedulingProvider.java
b/samza-rest/src/test/java/org/apache/samza/monitor/mock/InstantSchedulingProvider.java
index 6ae80e6..5c06911 100644
---
a/samza-rest/src/test/java/org/apache/samza/monitor/mock/InstantSchedulingProvider.java
+++
b/samza-rest/src/test/java/org/apache/samza/monitor/mock/InstantSchedulingProvider.java
@@ -25,10 +25,10 @@ import org.apache.samza.monitor.SchedulingProvider;
*/
public class InstantSchedulingProvider implements SchedulingProvider {
- public void schedule(Runnable runnableMonitor, int interval) {
- runnableMonitor.run();
- }
+ public void schedule(Runnable runnableMonitor, int interval) {
+ runnableMonitor.run();
+ }
- // Nothing to stop because no deferred task was started
- public void stop() {}
+ // Nothing to stop because no deferred task was started
+ public void stop() {}
}
diff --git
a/samza-rest/src/test/java/org/apache/samza/rest/resources/TestJobsResource.java
b/samza-rest/src/test/java/org/apache/samza/rest/resources/TestJobsResource.java
index d19a181..6e53d84 100644
---
a/samza-rest/src/test/java/org/apache/samza/rest/resources/TestJobsResource.java
+++
b/samza-rest/src/test/java/org/apache/samza/rest/resources/TestJobsResource.java
@@ -125,7 +125,7 @@ public class TestJobsResource extends BaseJerseyTest {
Response resp = target(String.format("v1/jobs/%s/%s", "BadJobName",
MockJobProxy.JOB_INSTANCE_2_ID)).request().get();
assertEquals(404, resp.getStatus());
- final Map<String, String> errorMessage =
objectMapper.readValue(resp.readEntity(String.class), new
TypeReference<Map<String, String>>() {});
+ final Map<String, String> errorMessage =
objectMapper.readValue(resp.readEntity(String.class), new
TypeReference<Map<String, String>>() { });
assertTrue(errorMessage.get("message"),
errorMessage.get("message").contains("does not exist"));
resp.close();
}
@@ -136,7 +136,7 @@ public class TestJobsResource extends BaseJerseyTest {
Response resp = target(String.format("v1/jobs/%s/%s",
MockJobProxy.JOB_INSTANCE_2_NAME, "BadJobId")).request().get();
assertEquals(404, resp.getStatus());
- final Map<String, String> errorMessage =
objectMapper.readValue(resp.readEntity(String.class), new
TypeReference<Map<String, String>>() {});
+ final Map<String, String> errorMessage =
objectMapper.readValue(resp.readEntity(String.class), new
TypeReference<Map<String, String>>() { });
assertTrue(errorMessage.get("message"),
errorMessage.get("message").contains("does not exist"));
resp.close();
}
@@ -184,7 +184,7 @@ public class TestJobsResource extends BaseJerseyTest {
.queryParam("status", "BADSTATUS").request().put(Entity.form(new
Form()));
assertEquals(400, resp.getStatus());
- final Map<String, String> errorMessage =
objectMapper.readValue(resp.readEntity(String.class), new
TypeReference<Map<String, String>>() {});
+ final Map<String, String> errorMessage =
objectMapper.readValue(resp.readEntity(String.class), new
TypeReference<Map<String, String>>() { });
assertTrue(errorMessage.get("message").contains("BADSTATUS"));
resp.close();
}
@@ -196,7 +196,7 @@ public class TestJobsResource extends BaseJerseyTest {
.put(Entity.form(new Form()));
assertEquals(400, resp.getStatus());
- final Map<String, String> errorMessage =
objectMapper.readValue(resp.readEntity(String.class), new
TypeReference<Map<String, String>>() {});
+ final Map<String, String> errorMessage =
objectMapper.readValue(resp.readEntity(String.class), new
TypeReference<Map<String, String>>() { });
assertTrue(errorMessage.get("message").contains("status"));
resp.close();
}
diff --git
a/samza-rest/src/test/java/org/apache/samza/rest/resources/TestTasksResource.java
b/samza-rest/src/test/java/org/apache/samza/rest/resources/TestTasksResource.java
index 7cb9527..ecb54ad 100644
---
a/samza-rest/src/test/java/org/apache/samza/rest/resources/TestTasksResource.java
+++
b/samza-rest/src/test/java/org/apache/samza/rest/resources/TestTasksResource.java
@@ -78,7 +78,7 @@ public class TestTasksResource extends BaseJerseyTest {
String requestUrl = String.format("v1/jobs/%s/%s/tasks", "BadJobName",
MockJobProxy.JOB_INSTANCE_4_ID);
Response resp = target(requestUrl).request().get();
assertEquals(400, resp.getStatus());
- final Map<String, String> errorMessage =
objectMapper.readValue(resp.readEntity(String.class), new
TypeReference<Map<String, String>>() {});
+ final Map<String, String> errorMessage =
objectMapper.readValue(resp.readEntity(String.class), new
TypeReference<Map<String, String>>() { });
assertTrue(errorMessage.get("message"),
errorMessage.get("message").contains("Invalid arguments for getTasks. "));
resp.close();
}
@@ -88,7 +88,7 @@ public class TestTasksResource extends BaseJerseyTest {
String requestUrl = String.format("v1/jobs/%s/%s/tasks",
MockJobProxy.JOB_INSTANCE_1_NAME, "BadJobId");
Response resp = target(requestUrl).request().get();
assertEquals(400, resp.getStatus());
- final Map<String, String> errorMessage =
objectMapper.readValue(resp.readEntity(String.class), new
TypeReference<Map<String, String>>() {});
+ final Map<String, String> errorMessage =
objectMapper.readValue(resp.readEntity(String.class), new
TypeReference<Map<String, String>>() { });
assertTrue(errorMessage.get("message"),
errorMessage.get("message").contains("Invalid arguments for getTasks. "));
resp.close();
}
diff --git
a/samza-rest/src/test/java/org/apache/samza/rest/resources/mock/MockJobProxyFactory.java
b/samza-rest/src/test/java/org/apache/samza/rest/resources/mock/MockJobProxyFactory.java
index 03e95b1..eeb681d 100644
---
a/samza-rest/src/test/java/org/apache/samza/rest/resources/mock/MockJobProxyFactory.java
+++
b/samza-rest/src/test/java/org/apache/samza/rest/resources/mock/MockJobProxyFactory.java
@@ -23,7 +23,7 @@ import org.apache.samza.rest.proxy.job.JobProxyFactory;
import org.apache.samza.rest.resources.JobsResourceConfig;
-public class MockJobProxyFactory implements JobProxyFactory{
+public class MockJobProxyFactory implements JobProxyFactory {
@Override
public JobProxy getJobProxy(JobsResourceConfig config) {
return new MockJobProxy(config);
diff --git
a/samza-rest/src/test/java/org/apache/samza/rest/resources/mock/MockJobStatusProvider.java
b/samza-rest/src/test/java/org/apache/samza/rest/resources/mock/MockJobStatusProvider.java
index df0f18a..d02b88f 100644
---
a/samza-rest/src/test/java/org/apache/samza/rest/resources/mock/MockJobStatusProvider.java
+++
b/samza-rest/src/test/java/org/apache/samza/rest/resources/mock/MockJobStatusProvider.java
@@ -30,9 +30,9 @@ public class MockJobStatusProvider implements
JobStatusProvider {
@Override
public void getJobStatuses(Collection<Job> jobs)
throws IOException, InterruptedException {
- for (Job info : jobs) {
- setStatusStarted(info);
- }
+ for (Job info : jobs) {
+ setStatusStarted(info);
+ }
}
@Override
diff --git
a/samza-yarn/src/main/java/org/apache/samza/job/yarn/FileSystemImplConfig.java
b/samza-yarn/src/main/java/org/apache/samza/job/yarn/FileSystemImplConfig.java
index 7e10f4f..a582dbb 100644
---
a/samza-yarn/src/main/java/org/apache/samza/job/yarn/FileSystemImplConfig.java
+++
b/samza-yarn/src/main/java/org/apache/samza/job/yarn/FileSystemImplConfig.java
@@ -20,11 +20,7 @@ package org.apache.samza.job.yarn;
import java.util.ArrayList;
import java.util.List;
-import java.util.Set;
-import org.apache.commons.lang.StringUtils;
import org.apache.samza.config.Config;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
diff --git
a/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnAppState.java
b/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnAppState.java
index 4d15b1a..15ca92f 100644
--- a/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnAppState.java
+++ b/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnAppState.java
@@ -22,7 +22,6 @@ package org.apache.samza.job.yarn;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
-import org.apache.samza.job.yarn.YarnContainer;
import java.net.URL;
import java.util.Map;
diff --git
a/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
b/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
index 5459a9d..46507d9 100644
---
a/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
+++
b/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
@@ -81,7 +81,7 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
private static final int PREFERRED_HOST_PRIORITY = 0;
private static final int ANY_HOST_PRIORITY = 1;
- private final String INVALID_PROCESSOR_ID = "-1";
+ private static final String INVALID_PROCESSOR_ID = "-1";
/**
* The AMClient instance to request resources from yarn.
@@ -153,11 +153,11 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
// Use the Samza job config "fs.<scheme>.impl" and "fs.<scheme>.impl.*"
for YarnConfiguration
FileSystemImplConfig fsImplConfig = new FileSystemImplConfig(config);
fsImplConfig.getSchemes().forEach(
- scheme -> {
- fsImplConfig.getSchemeConfig(scheme).forEach(
- (confKey, confValue) -> yarnConfiguration.set(confKey, confValue)
- );
- }
+ scheme -> {
+ fsImplConfig.getSchemeConfig(scheme).forEach(
+ (confKey, confValue) -> yarnConfiguration.set(confKey, confValue)
+ );
+ }
);
MetricsRegistryMap registry = new MetricsRegistryMap();
@@ -204,7 +204,7 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
*/
@Override
public void start() {
- if(!started.compareAndSet(false, true)) {
+ if (!started.compareAndSet(false, true)) {
log.info("Attempting to start an already started
YarnClusterResourceManager");
return;
}
@@ -217,7 +217,7 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
nmClientAsync.start();
lifecycle.onInit();
- if(lifecycle.shouldShutdown()) {
+ if (lifecycle.shouldShutdown()) {
clusterManagerCallback.onError(new SamzaException("Invalid resource
request."));
}
@@ -338,11 +338,11 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
//In that case, this scan will turn into a lookup. This change will require
changes/testing in the UI files because
//those UI stub templates operate on the YarnContainer object.
private String getRunningProcessorId(String containerId) {
- for(Map.Entry<String, YarnContainer> entry :
state.runningProcessors.entrySet()) {
+ for (Map.Entry<String, YarnContainer> entry :
state.runningProcessors.entrySet()) {
String key = entry.getKey();
YarnContainer yarnContainer = entry.getValue();
String yarnContainerId = yarnContainer.id().toString();
- if(yarnContainerId.equals(containerId)) {
+ if (yarnContainerId.equals(containerId)) {
return key;
}
}
@@ -395,7 +395,7 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
service.onShutdown();
metrics.stop();
- if(status != SamzaApplicationState.SamzaAppStatus.UNDEFINED) {
+ if (status != SamzaApplicationState.SamzaAppStatus.UNDEFINED) {
cleanupStagingDir();
}
}
@@ -406,7 +406,7 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
*/
private void cleanupStagingDir() {
String yarnJobStagingDirectory = yarnConfig.getYarnJobStagingDirectory();
- if(yarnJobStagingDirectory != null) {
+ if (yarnJobStagingDirectory != null) {
JobContext context = new JobContext();
context.setAppStagingDir(new Path(yarnJobStagingDirectory));
@@ -417,7 +417,7 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
log.error("Unable to clean up file system.", e);
return;
}
- if(fs != null) {
+ if (fs != null) {
YarnJobUtil.cleanupStagingDir(context, fs);
}
}
@@ -433,7 +433,7 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
public void onContainersCompleted(List<ContainerStatus> statuses) {
List<SamzaResourceStatus> samzaResourceStatuses = new ArrayList<>();
- for(ContainerStatus status: statuses) {
+ for (ContainerStatus status : statuses) {
log.info("Got completion notification for Container ID: {} with status:
{} and state: {}. Diagnostics information: {}.",
status.getContainerId(), status.getExitStatus(), status.getState(),
status.getDiagnostics());
@@ -445,12 +445,12 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
//remove the container from the list of running containers, if failed
with a non-zero exit code, add it to the list of
//failed containers.
- if(!completedProcessorID.equals(INVALID_PROCESSOR_ID)){
- if(state.runningProcessors.containsKey(completedProcessorID)) {
+ if (!completedProcessorID.equals(INVALID_PROCESSOR_ID)) {
+ if (state.runningProcessors.containsKey(completedProcessorID)) {
log.info("Removing Processor ID: {} from YarnClusterResourceManager
running processors.", completedProcessorID);
state.runningProcessors.remove(completedProcessorID);
- if(status.getExitStatus() != ContainerExitStatus.SUCCESS)
+ if (status.getExitStatus() != ContainerExitStatus.SUCCESS)
state.failedContainersStatus.put(status.getContainerId().toString(), status);
}
}
@@ -465,19 +465,20 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
*/
@Override
public void onContainersAllocated(List<Container> containers) {
- List<SamzaResource> resources = new ArrayList<SamzaResource>();
- for(Container container : containers) {
- log.info("Got allocation notification for Container ID: {} on host:
{}", container.getId(), container.getNodeId().getHost());
- String containerId = container.getId().toString();
- String host = container.getNodeId().getHost();
- int memory = container.getResource().getMemory();
- int numCores = container.getResource().getVirtualCores();
-
- SamzaResource resource = new SamzaResource(numCores, memory, host,
containerId);
- allocatedResources.put(resource, container);
- resources.add(resource);
- }
- clusterManagerCallback.onResourcesAvailable(resources);
+ List<SamzaResource> resources = new ArrayList<SamzaResource>();
+ for (Container container : containers) {
+ log.info("Got allocation notification for Container ID: {} on host: {}",
container.getId(),
+ container.getNodeId().getHost());
+ String containerId = container.getId().toString();
+ String host = container.getNodeId().getHost();
+ int memory = container.getResource().getMemory();
+ int numCores = container.getResource().getVirtualCores();
+
+ SamzaResource resource = new SamzaResource(numCores, memory, host,
containerId);
+ allocatedResources.put(resource, container);
+ resources.add(resource);
+ }
+ clusterManagerCallback.onResourcesAvailable(resources);
}
//The below methods are specific to the Yarn AMRM Client. We currently don't
handle scenarios where there are
@@ -513,7 +514,7 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
public void onContainerStarted(ContainerId containerId, Map<String,
ByteBuffer> allServiceResponse) {
String processorId = getPendingProcessorId(containerId);
if (processorId != null) {
- log.info("Got start notification for Container ID: {} for Processor ID:
{}", containerId, processorId);
+ log.info("Got start notification for Container ID: {} for Processor ID:
{}", containerId, processorId);
// 1. Move the processor from pending to running state
final YarnContainer container =
state.pendingProcessors.remove(processorId);
@@ -664,7 +665,11 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
ContainerLaunchContext context =
Records.newRecord(ContainerLaunchContext.class);
context.setEnvironment(env);
context.setTokens(allTokens.duplicate());
- context.setCommands(new ArrayList<String>() {{add(cmd);}});
+ context.setCommands(new ArrayList<String>() {
+ {
+ add(cmd);
+ }
+ });
context.setLocalResources(localResourceMap);
if (UserGroupInformation.isSecurityEnabled()) {
diff --git
a/samza-yarn/src/main/java/org/apache/samza/validation/YarnJobValidationTool.java
b/samza-yarn/src/main/java/org/apache/samza/validation/YarnJobValidationTool.java
index 23f0e4d..2b31977 100644
---
a/samza-yarn/src/main/java/org/apache/samza/validation/YarnJobValidationTool.java
+++
b/samza-yarn/src/main/java/org/apache/samza/validation/YarnJobValidationTool.java
@@ -93,7 +93,7 @@ public class YarnJobValidationTool {
appId = validateAppId();
attemptId = validateRunningAttemptId(appId);
validateContainerCount(attemptId);
- if(validator != null) {
+ if (validator != null) {
validateJmxMetrics();
}
@@ -108,10 +108,10 @@ public class YarnJobValidationTool {
// fetch only the last created application with the job name and id
// i.e. get the application with max appId
ApplicationId appId = null;
- for(ApplicationReport applicationReport : this.client.getApplications()) {
- if(applicationReport.getName().equals(this.jobName)) {
+ for (ApplicationReport applicationReport : this.client.getApplications()) {
+ if (applicationReport.getName().equals(this.jobName)) {
ApplicationId id = applicationReport.getApplicationId();
- if(appId == null || appId.compareTo(id) < 0) {
+ if (appId == null || appId.compareTo(id) < 0) {
appId = id;
}
}
@@ -137,8 +137,8 @@ public class YarnJobValidationTool {
public int validateContainerCount(ApplicationAttemptId attemptId) throws
Exception {
int runningContainerCount = 0;
- for(ContainerReport containerReport :
this.client.getContainers(attemptId)) {
- if(containerReport.getContainerState() == ContainerState.RUNNING) {
+ for (ContainerReport containerReport :
this.client.getContainers(attemptId)) {
+ if (containerReport.getContainerState() == ContainerState.RUNNING) {
++runningContainerCount;
}
}
@@ -157,7 +157,7 @@ public class YarnJobValidationTool {
MetricsRegistry metricsRegistry = new MetricsRegistryMap();
CoordinatorStreamStore coordinatorStreamStore = new
CoordinatorStreamStore(config, metricsRegistry);
coordinatorStreamStore.init();
- try{
+ try {
Config configFromCoordinatorStream =
CoordinatorStreamUtil.readConfigFromCoordinatorStream(coordinatorStreamStore);
ChangelogStreamManager changelogStreamManager = new
ChangelogStreamManager(coordinatorStreamStore);
JobModelManager jobModelManager =
@@ -181,7 +181,7 @@ public class YarnJobValidationTool {
}
}
- public static void main(String [] args) throws Exception {
+ public static void main(String[] args) throws Exception {
CommandLine cmdline = new CommandLine();
OptionParser parser = cmdline.parser();
OptionSpec<String> validatorOpt = parser.accepts("metrics-validator", "The
metrics validator class.")
diff --git
a/samza-yarn/src/main/java/org/apache/samza/webapp/ApplicationMasterRestClient.java
b/samza-yarn/src/main/java/org/apache/samza/webapp/ApplicationMasterRestClient.java
index eed16db..354e71e 100644
---
a/samza-yarn/src/main/java/org/apache/samza/webapp/ApplicationMasterRestClient.java
+++
b/samza-yarn/src/main/java/org/apache/samza/webapp/ApplicationMasterRestClient.java
@@ -53,7 +53,7 @@ public class ApplicationMasterRestClient implements Closeable
{
*/
public Map<String, Map<String, Object>> getMetrics() throws IOException {
String jsonString = getEntityAsJson("/metrics", "metrics");
- return jsonMapper.readValue(jsonString, new TypeReference<Map<String,
Map<String, Object>>>() {});
+ return jsonMapper.readValue(jsonString, new TypeReference<Map<String,
Map<String, Object>>>() { });
}
/**
@@ -62,7 +62,7 @@ public class ApplicationMasterRestClient implements Closeable
{
*/
public Map<String, Object> getTaskContext() throws IOException {
String jsonString = getEntityAsJson("/task-context", "task context");
- return jsonMapper.readValue(jsonString, new TypeReference<Map<String,
Object>>() {});
+ return jsonMapper.readValue(jsonString, new TypeReference<Map<String,
Object>>() { });
}
/**
@@ -71,7 +71,7 @@ public class ApplicationMasterRestClient implements Closeable
{
*/
public Map<String, Object> getAmState() throws IOException {
String jsonString = getEntityAsJson("/am", "AM state");
- return jsonMapper.readValue(jsonString, new TypeReference<Map<String,
Object>>() {});
+ return jsonMapper.readValue(jsonString, new TypeReference<Map<String,
Object>>() { });
}
/**
@@ -80,7 +80,7 @@ public class ApplicationMasterRestClient implements Closeable
{
*/
public Map<String, Object> getConfig() throws IOException {
String jsonString = getEntityAsJson("/config", "config");
- return jsonMapper.readValue(jsonString, new TypeReference<Map<String,
Object>>() {});
+ return jsonMapper.readValue(jsonString, new TypeReference<Map<String,
Object>>() { });
}
@Override
diff --git
a/samza-yarn/src/test/java/org/apache/samza/config/TestYarnConfig.java
b/samza-yarn/src/test/java/org/apache/samza/config/TestYarnConfig.java
index 1018a75..ae4bf25 100644
--- a/samza-yarn/src/test/java/org/apache/samza/config/TestYarnConfig.java
+++ b/samza-yarn/src/test/java/org/apache/samza/config/TestYarnConfig.java
@@ -19,12 +19,10 @@
package org.apache.samza.config;
import java.util.Collections;
-import java.util.Optional;
import org.apache.samza.SamzaException;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
public class TestYarnConfig {
diff --git
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestFileSystemImplConfig.java
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestFileSystemImplConfig.java
index 63d83df..b906efd 100644
---
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestFileSystemImplConfig.java
+++
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestFileSystemImplConfig.java
@@ -67,13 +67,13 @@ public class TestFileSystemImplConfig {
FileSystemImplConfig manager = new FileSystemImplConfig(conf);
- Map<String, String> expectedFsHttpImplConfs = ImmutableMap.of( //Scheme
with additional subkeys
+ Map<String, String> expectedFsHttpImplConfs = ImmutableMap.of(//Scheme
with additional subkeys
"fs.http.impl", "org.apache.samza.HttpFileSystem",
"fs.http.impl.key1", "val1",
"fs.http.impl.key2", "val2"
);
- Map<String, String> expectedFsMyschemeImplConfs = ImmutableMap.of( //
Scheme without subkeys
+ Map<String, String> expectedFsMyschemeImplConfs = ImmutableMap.of(//
Scheme without subkeys
"fs.myscheme.impl", "org.apache.samza.MySchemeFileSystem"
);
diff --git
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestLocalizerResourceConfig.java
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestLocalizerResourceConfig.java
index e003125..6d0b8af 100644
---
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestLocalizerResourceConfig.java
+++
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestLocalizerResourceConfig.java
@@ -32,7 +32,7 @@ import static org.junit.Assert.assertEquals;
public class TestLocalizerResourceConfig {
@Rule
- public ExpectedException thrown= ExpectedException.none();
+ public ExpectedException thrown = ExpectedException.none();
@Test
public void testResourceConfigIncluded() {
diff --git
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestLocalizerResourceMapper.java
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestLocalizerResourceMapper.java
index d065019..6ff8df1 100644
---
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestLocalizerResourceMapper.java
+++
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestLocalizerResourceMapper.java
@@ -37,7 +37,7 @@ import static org.junit.Assert.assertNull;
public class TestLocalizerResourceMapper {
@Rule
- public ExpectedException thrown= ExpectedException.none();
+ public ExpectedException thrown = ExpectedException.none();
@Test
public void testResourceMapSuccess() {
diff --git
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestYarnClusterResourceManager.java
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestYarnClusterResourceManager.java
index 8f19eab..0898800 100644
---
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestYarnClusterResourceManager.java
+++
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestYarnClusterResourceManager.java
@@ -65,7 +65,7 @@ public class TestYarnClusterResourceManager {
new YarnContainer(Container.newInstance(
ContainerId.newContainerId(
ApplicationAttemptId.newInstance(
- ApplicationId.newInstance(10000l, 1), 1), 1),
+ ApplicationId.newInstance(10000L, 1), 1), 1),
NodeId.newInstance("host1", 8088), "http://host1";,
Resource.newInstance(1024, 1), Priority.newInstance(1),
Token.newInstance("id".getBytes(), "read", "password".getBytes(),
"service"))));
@@ -75,7 +75,7 @@ public class TestYarnClusterResourceManager {
yarnClusterResourceManager.onStartContainerError(ContainerId.newContainerId(
ApplicationAttemptId.newInstance(
- ApplicationId.newInstance(10000l, 1), 1), 1),
+ ApplicationId.newInstance(10000L, 1), 1), 1),
new Exception());
assertEquals(0, yarnAppState.pendingProcessors.size());
diff --git
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestYarnJobFactory.java
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestYarnJobFactory.java
index 12d45f5..9ba6232 100644
--- a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestYarnJobFactory.java
+++ b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestYarnJobFactory.java
@@ -40,7 +40,7 @@ public class TestYarnJobFactory {
YarnJobFactory jobFactory = new YarnJobFactory();
YarnJob yarnJob = jobFactory.getJob(new MapConfig(ImmutableMap.of(
"fs.http.impl", "org.apache.myHttp",
- "fs.myscheme.impl","org.apache.myScheme")));
+ "fs.myscheme.impl", "org.apache.myScheme")));
Configuration hConfig = yarnJob.client().yarnClient().getConfig();
assertEquals("org.apache.myHttp", hConfig.get("fs.http.impl"));
assertEquals("org.apache.myScheme", hConfig.get("fs.myscheme.impl"));
@@ -50,8 +50,8 @@ public class TestYarnJobFactory {
public void testGetJobWithFsImplSubkeys() {
YarnJobFactory jobFactory = new YarnJobFactory();
YarnJob yarnJob = jobFactory.getJob(new MapConfig(ImmutableMap.of(
- "fs.myscheme.impl","org.apache.myScheme",
- "fs.myscheme.impl.client","org.apache.mySchemeClient")));
+ "fs.myscheme.impl", "org.apache.myScheme",
+ "fs.myscheme.impl.client", "org.apache.mySchemeClient")));
Configuration hConfig = yarnJob.client().yarnClient().getConfig();
assertEquals("org.apache.myScheme", hConfig.get("fs.myscheme.impl"));
assertEquals("org.apache.mySchemeClient",
hConfig.get("fs.myscheme.impl.client"));
diff --git
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/util/MockContainerListener.java
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/util/MockContainerListener.java
index 43bda8f..69c9745 100644
---
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/util/MockContainerListener.java
+++
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/util/MockContainerListener.java
@@ -21,7 +21,6 @@ package org.apache.samza.job.yarn.util;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
-import org.apache.hadoop.yarn.api.records.Container;
import static org.junit.Assert.assertTrue;
diff --git
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/util/MockHttpServer.java
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/util/MockHttpServer.java
index a6a31d7..7435f33 100644
---
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/util/MockHttpServer.java
+++
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/util/MockHttpServer.java
@@ -43,7 +43,7 @@ public class MockHttpServer extends HttpServer {
@Override
public URL getUrl() {
- if(running()) {
+ if (running()) {
try {
return new URL("http://localhost:12345/";);
} catch (MalformedURLException mue) {
diff --git
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/util/hadoop/TestHttpFileSystem.java
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/util/hadoop/TestHttpFileSystem.java
index 6f42856..53f8c40 100644
---
a/samza-yarn/src/test/java/org/apache/samza/job/yarn/util/hadoop/TestHttpFileSystem.java
+++
b/samza-yarn/src/test/java/org/apache/samza/job/yarn/util/hadoop/TestHttpFileSystem.java
@@ -85,13 +85,13 @@ public class TestHttpFileSystem {
//Hang the connection until the read timeout expires on the client
side.
if (numBytesWritten >= THRESHOLD_BYTES) {
- if(!serverWaitLatch.await(5, TimeUnit.SECONDS)) {
+ if (!serverWaitLatch.await(5, TimeUnit.SECONDS)) {
throw new IOException("Timed out waiting for latch");
}
break;
}
}
- } catch(Exception e) {
+ } catch (Exception e) {
//Record any exception that may have occurred
LOG.error("{}", e);
serverException = e;
@@ -129,10 +129,10 @@ public class TestHttpFileSystem {
while (in.read() >= 0) {
totalBytesRead++;
}
- } catch(SocketTimeoutException e) {
+ } catch (SocketTimeoutException e) {
//Expect the socket to timeout after THRESHOLD bytes have been read.
serverWaitLatch.countDown();
- } catch(Exception e) {
+ } catch (Exception e) {
//Record any exception that may have occurred.
LOG.error("{}", e);
clientException = e;
diff --git
a/samza-yarn/src/test/java/org/apache/samza/validation/MockMetricsValidator.java
b/samza-yarn/src/test/java/org/apache/samza/validation/MockMetricsValidator.java
index c3cf935..de0a980 100644
---
a/samza-yarn/src/test/java/org/apache/samza/validation/MockMetricsValidator.java
+++
b/samza-yarn/src/test/java/org/apache/samza/validation/MockMetricsValidator.java
@@ -36,9 +36,11 @@ public class MockMetricsValidator implements
MetricsValidator {
@Override
public void validate(MetricsAccessor accessor) throws
MetricsValidationFailureException {
Map<String, Long> commitCalls =
accessor.getCounterValues(SamzaContainerMetrics.class.getName(),
"commit-calls");
- if(commitCalls.isEmpty()) throw new MetricsValidationFailureException("no
value");
- for(Map.Entry<String, Long> entry: commitCalls.entrySet()) {
- if(entry.getValue() <= 0) {
+ if (commitCalls.isEmpty()) {
+ throw new MetricsValidationFailureException("no value");
+ }
+ for (Map.Entry<String, Long> entry : commitCalls.entrySet()) {
+ if (entry.getValue() <= 0) {
throw new MetricsValidationFailureException("commit call <= 0");
}
}
diff --git
a/samza-yarn/src/test/java/org/apache/samza/webapp/TestApplicationMasterRestClient.java
b/samza-yarn/src/test/java/org/apache/samza/webapp/TestApplicationMasterRestClient.java
index 0dbbed0..338b455 100644
---
a/samza-yarn/src/test/java/org/apache/samza/webapp/TestApplicationMasterRestClient.java
+++
b/samza-yarn/src/test/java/org/apache/samza/webapp/TestApplicationMasterRestClient.java
@@ -300,7 +300,7 @@ public class TestApplicationMasterRestClient {
containers.put(yarnContainerId, containerMap);
});
- return jsonMapper.readValue(jsonMapper.writeValueAsString(containers), new
TypeReference<Map<String, Map<String, Object>>>() {});
+ return jsonMapper.readValue(jsonMapper.writeValueAsString(containers), new
TypeReference<Map<String, Map<String, Object>>>() { });
}
private void assignMetricValues(SamzaApplicationState samzaAppState,
MetricsRegistryMap registry) {
diff --git
a/samza-yarn/src/test/java/org/apache/samza/webapp/TestYarnContainerHeartbeatServlet.java
b/samza-yarn/src/test/java/org/apache/samza/webapp/TestYarnContainerHeartbeatServlet.java
index f7ec995..8987834 100644
---
a/samza-yarn/src/test/java/org/apache/samza/webapp/TestYarnContainerHeartbeatServlet.java
+++
b/samza-yarn/src/test/java/org/apache/samza/webapp/TestYarnContainerHeartbeatServlet.java
@@ -73,10 +73,10 @@ public class TestYarnContainerHeartbeatServlet {
@Test
public void testContainerHeartbeatWhenValid()
throws IOException {
- String VALID_CONTAINER_ID = "container_1350670447861_0003_01_000002";
-
when(container.id()).thenReturn(ConverterUtils.toContainerId(VALID_CONTAINER_ID));
- yarnAppState.runningProcessors.put(VALID_CONTAINER_ID, container);
- URL url = new URL(webApp.getUrl().toString() +
"containerHeartbeat?executionContainerId=" + VALID_CONTAINER_ID);
+ String validContainerId = "container_1350670447861_0003_01_000002";
+
when(container.id()).thenReturn(ConverterUtils.toContainerId(validContainerId));
+ yarnAppState.runningProcessors.put(validContainerId, container);
+ URL url = new URL(webApp.getUrl().toString() +
"containerHeartbeat?executionContainerId=" + validContainerId);
String response = HttpUtil.read(url, 1000, new ExponentialSleepStrategy());
heartbeat = mapper.readValue(response, ContainerHeartbeatResponse.class);
Assert.assertTrue(heartbeat.isAlive());
@@ -85,11 +85,11 @@ public class TestYarnContainerHeartbeatServlet {
@Test
public void testContainerHeartbeatWhenInvalid()
throws IOException {
- String VALID_CONTAINER_ID = "container_1350670447861_0003_01_000003";
- String INVALID_CONTAINER_ID = "container_1350670447861_0003_01_000002";
-
when(container.id()).thenReturn(ConverterUtils.toContainerId(VALID_CONTAINER_ID));
- yarnAppState.runningProcessors.put(VALID_CONTAINER_ID, container);
- URL url = new URL(webApp.getUrl().toString() +
"containerHeartbeat?executionContainerId=" + INVALID_CONTAINER_ID);
+ String validContainerId = "container_1350670447861_0003_01_000003";
+ String invalidContainerId = "container_1350670447861_0003_01_000002";
+
when(container.id()).thenReturn(ConverterUtils.toContainerId(validContainerId));
+ yarnAppState.runningProcessors.put(validContainerId, container);
+ URL url = new URL(webApp.getUrl().toString() +
"containerHeartbeat?executionContainerId=" + invalidContainerId);
String response = HttpUtil.read(url, 1000, new ExponentialSleepStrategy());
heartbeat = mapper.readValue(response, ContainerHeartbeatResponse.class);
Assert.assertFalse(heartbeat.isAlive());
