diff --git a/build.gradle b/build.gradle
index b7d28983..c1dc8597 100644
--- a/build.gradle
+++ b/build.gradle
@@ -89,6 +89,8 @@ subprojects {
apply plugin: "license"
apply plugin: "com.google.protobuf"
apply plugin: "java"
+ apply plugin: "checkstyle"
+ apply plugin: "findbugs"
apply plugin: "eclipse"
apply plugin: "idea"
apply plugin: "maven-publish"
@@ -180,6 +182,26 @@ subprojects {
signing {
sign publishing.publications.mavenJava
}
+
+ checkstyle {
+ configFile rootProject.file('config/checkstyle/checkstyle.xml')
+ maxWarnings 0
+ }
+
+ tasks.withType(FindBugs) {
+ reports {
+ xml.enabled false
+ html.enabled true
+ }
+
+ classpath = classpath.filter {
+ // POM files are getting included in the classpath for some reason, but this causes parsing errors when FindBugs
+ // tries to analyze them, thinking they are zips. Excluding POM files as a workaround.
+ !it.name.endsWith('.pom')
+ }
+
+ excludeFilter = rootProject.file('config/findbugs/findbugsExclude.xml')
+ }
}
apply plugin: 'distribution'
diff --git a/config/checkstyle/checkstyle.xml b/config/checkstyle/checkstyle.xml
new file mode 100644
index 00000000..a9bffe83
--- /dev/null
+++ b/config/checkstyle/checkstyle.xml
@@ -0,0 +1,198 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/config/checkstyle/suppressions.xml b/config/checkstyle/suppressions.xml
new file mode 100644
index 00000000..253f46d4
--- /dev/null
+++ b/config/checkstyle/suppressions.xml
@@ -0,0 +1,7 @@
+
+
+
+
+
diff --git a/config/findbugs/findbugsExclude.xml b/config/findbugs/findbugsExclude.xml
new file mode 100644
index 00000000..3b6da0f1
--- /dev/null
+++ b/config/findbugs/findbugsExclude.xml
@@ -0,0 +1,5 @@
+
+
+
+
+
diff --git a/tony-azkaban/src/main/java/com/linkedin/tony/azkaban/TensorFlowJob.java b/tony-azkaban/src/main/java/com/linkedin/tony/azkaban/TensorFlowJob.java
index 82069110..cf889f04 100644
--- a/tony-azkaban/src/main/java/com/linkedin/tony/azkaban/TensorFlowJob.java
+++ b/tony-azkaban/src/main/java/com/linkedin/tony/azkaban/TensorFlowJob.java
@@ -84,40 +84,41 @@ protected String getJVMArguments() {
@Override
protected String getMainArguments() {
- String args = super.getMainArguments();
+ StringBuilder args = new StringBuilder(super.getMainArguments());
info("All job props: " + getJobProps());
info("All sys props: " + getSysProps());
String srcDir = getJobProps().getString(TensorFlowJobArg.SRC_DIR.azPropName, "src");
- args += " " + TensorFlowJobArg.SRC_DIR.tonyParamName + " " + srcDir;
+ args.append(" " + TensorFlowJobArg.SRC_DIR.tonyParamName + " " + srcDir);
String hdfsClasspath = getJobProps().getString(TensorFlowJobArg.HDFS_CLASSPATH.azPropName, null);
if (hdfsClasspath != null) {
- args += " " + TensorFlowJobArg.HDFS_CLASSPATH.tonyParamName + " " + hdfsClasspath;
+ args.append(" " + TensorFlowJobArg.HDFS_CLASSPATH.tonyParamName + " " + hdfsClasspath);
}
Map workerEnvs = getJobProps().getMapByPrefix(WORKER_ENV_PREFIX);
for (Map.Entry workerEnv : workerEnvs.entrySet()) {
- args += " " + TensorFlowJobArg.SHELL_ENV.tonyParamName + " " + workerEnv.getKey() + "=" + workerEnv.getValue();
+ args.append(" " + TensorFlowJobArg.SHELL_ENV.tonyParamName + " " + workerEnv.getKey()
+ + "=" + workerEnv.getValue());
}
String taskParams = getJobProps().getString(TensorFlowJobArg.TASK_PARAMS.azPropName, null);
if (taskParams != null) {
- args += " " + TensorFlowJobArg.TASK_PARAMS.tonyParamName + " '" + taskParams + "'";
+ args.append(" " + TensorFlowJobArg.TASK_PARAMS.tonyParamName + " '" + taskParams + "'");
}
String pythonBinaryPath = getJobProps().getString(TensorFlowJobArg.PYTHON_BINARY_PATH.azPropName, null);
if (pythonBinaryPath != null) {
- args += " " + TensorFlowJobArg.PYTHON_BINARY_PATH.tonyParamName + " " + pythonBinaryPath;
+ args.append(" " + TensorFlowJobArg.PYTHON_BINARY_PATH.tonyParamName + " " + pythonBinaryPath);
}
String pythonVenv = getJobProps().getString(TensorFlowJobArg.PYTHON_VENV.azPropName, null);
if (pythonVenv != null) {
- args += " " + TensorFlowJobArg.PYTHON_VENV.tonyParamName + " " + pythonVenv;
+ args.append(" " + TensorFlowJobArg.PYTHON_VENV.tonyParamName + " " + pythonVenv);
}
String executes = getJobProps().getString(TensorFlowJobArg.EXECUTES.azPropName, null);
if (executes != null) {
- args += " " + TensorFlowJobArg.EXECUTES.tonyParamName + " " + executes;
+ args.append(" " + TensorFlowJobArg.EXECUTES.tonyParamName + " " + executes);
}
Map tonyConfs = getJobProps().getMapByPrefix(TONY_CONF_PREFIX);
@@ -127,7 +128,9 @@ protected String getMainArguments() {
}
// Write user's tony confs to an xml to be localized.
- tonyConfFile.getParentFile().mkdir();
+ if (!tonyConfFile.getParentFile().mkdir()) {
+ throw new RuntimeException("Failed to create parent directory for TonY conf file.");
+ }
try (OutputStream os = new FileOutputStream(tonyConfFile)) {
tfConf.writeXml(os);
} catch (Exception e) {
@@ -136,6 +139,6 @@ protected String getMainArguments() {
info("Complete main arguments: " + args);
- return args;
+ return args.toString();
}
}
diff --git a/tony-azkaban/src/test/java/com/linkedin/tony/azkaban/TestTensorFlowJob.java b/tony-azkaban/src/test/java/com/linkedin/tony/azkaban/TestTensorFlowJob.java
index 848ae029..2a098cfd 100644
--- a/tony-azkaban/src/test/java/com/linkedin/tony/azkaban/TestTensorFlowJob.java
+++ b/tony-azkaban/src/test/java/com/linkedin/tony/azkaban/TestTensorFlowJob.java
@@ -29,7 +29,7 @@ public class TestTensorFlowJob {
private static void initServiceProvider() {
final Injector injector = Guice.createInjector(new AbstractModule() {
@Override
- protected void configure() {}
+ protected void configure() { }
});
// Because SERVICE_PROVIDER is a singleton and it is shared among many tests,
// need to reset the state to avoid assertion failures.
diff --git a/tony-cli/src/main/java/com/linkedin/tony/cli/NotebookSubmitter.java b/tony-cli/src/main/java/com/linkedin/tony/cli/NotebookSubmitter.java
index 21183afa..6e3d2825 100644
--- a/tony-cli/src/main/java/com/linkedin/tony/cli/NotebookSubmitter.java
+++ b/tony-cli/src/main/java/com/linkedin/tony/cli/NotebookSubmitter.java
@@ -58,7 +58,7 @@ public int submit(String[] args) throws ParseException, URISyntaxException, IOEx
opts.addOption("src_dir", true, "Name of directory of source files.");
int exitCode = 0;
- Path cachedLibPath = null;
+ Path cachedLibPath;
Configuration hdfsConf = new Configuration();
try (FileSystem fs = FileSystem.get(hdfsConf)) {
cachedLibPath = new Path(fs.getHomeDirectory(), TONY_FOLDER + Path.SEPARATOR + UUID.randomUUID().toString());
@@ -67,10 +67,7 @@ public int submit(String[] args) throws ParseException, URISyntaxException, IOEx
LOG.info("Copying " + jarPath + " to: " + cachedLibPath);
} catch (IOException e) {
LOG.fatal("Failed to create FileSystem: ", e);
- exitCode = -1;
- }
- if (cachedLibPath == null) {
- System.exit(-1);
+ return -1;
}
String[] updatedArgs = Arrays.copyOf(args, args.length + 4);
@@ -81,9 +78,6 @@ public int submit(String[] args) throws ParseException, URISyntaxException, IOEx
client.init(updatedArgs);
client.start();
- if (client == null) {
- System.exit(-1);
- }
Thread clientThread = new Thread(client::start);
clientThread.start();
while (clientThread.isAlive()) {
@@ -118,6 +112,5 @@ public static void main(String[] args) throws Exception {
NotebookSubmitter submitter = new NotebookSubmitter();
exitCode = submitter.submit(args);
System.exit(exitCode);
-
}
}
diff --git a/tony-core/src/main/java/com/linkedin/tony/TFConfig.java b/tony-core/src/main/java/com/linkedin/tony/TFConfig.java
index fd5367d7..4158cc62 100644
--- a/tony-core/src/main/java/com/linkedin/tony/TFConfig.java
+++ b/tony-core/src/main/java/com/linkedin/tony/TFConfig.java
@@ -20,7 +20,7 @@ public static class Task {
private int index;
// Jackson needs a default constructor
- Task() {}
+ Task() { }
Task(String type, int index) {
this.type = type;
@@ -47,7 +47,7 @@ public void setIndex(int index) {
}
// Jackson needs a default constructor
- TFConfig() {}
+ TFConfig() { }
public TFConfig(Map> clusterSpec, String jobName, int taskIndex) {
this.clusterSpec = clusterSpec;
diff --git a/tony-core/src/main/java/com/linkedin/tony/TFPolicyProvider.java b/tony-core/src/main/java/com/linkedin/tony/TFPolicyProvider.java
index 14b6583b..aaef4914 100644
--- a/tony-core/src/main/java/com/linkedin/tony/TFPolicyProvider.java
+++ b/tony-core/src/main/java/com/linkedin/tony/TFPolicyProvider.java
@@ -9,18 +9,11 @@
import com.linkedin.tony.rpc.TensorFlowCluster;
/**
- * * PolicyProvider for Client to AM protocol.
- * */
+ * PolicyProvider for Client to AM protocol.
+ **/
public class TFPolicyProvider extends PolicyProvider {
-
- private static final Service[] TF_AM_SERVICE =
- new Service[]{
- new Service(
- "security.tf.client-am-protocol.acl",
- TensorFlowCluster.class)};
-
@Override
public Service[] getServices() {
- return TF_AM_SERVICE;
- };
+ return new Service[]{new Service("security.tf.client-am-protocol.acl", TensorFlowCluster.class)};
+ }
}
diff --git a/tony-core/src/main/java/com/linkedin/tony/TaskExecutor.java b/tony-core/src/main/java/com/linkedin/tony/TaskExecutor.java
index bdcf2307..6b7e5522 100644
--- a/tony-core/src/main/java/com/linkedin/tony/TaskExecutor.java
+++ b/tony-core/src/main/java/com/linkedin/tony/TaskExecutor.java
@@ -165,6 +165,8 @@ public static void main(String[] args) throws Exception {
executor.shellEnv.put(Constants.WORLD, String.valueOf(executor.numTasks));
break;
}
+ default:
+ throw new RuntimeException("Unsupported executor framework: " + executor.framework);
}
int exitCode = Utils.executeShell(executor.taskCommand, executor.timeOut, executor.shellEnv);
@@ -210,8 +212,6 @@ private String registerAndGetClusterSpec(String amAddress) {
String hostName = Utils.getCurrentHostName();
LOG.info("ContainerId is: " + containerId + " HostName is: " + hostName);
- hangIfTesting();
-
// Start the Heartbeater..
hbExec.scheduleAtFixedRate(new Heartbeater(),
0, hbInterval, TimeUnit.MILLISECONDS);
@@ -273,9 +273,10 @@ public void run() {
} catch (Exception e) {
LOG.error("[" + taskId + "] Failed to send Heart Beat.", e);
if (++numFailedHBAttempts > MAX_NUM_FAILED_HB_ATTEMPTS) {
- LOG.error("[" + taskId + "] Exceeded Failed Heart Beat send attempts.. going to die !!");
+ LOG.error("[" + taskId + "] Exceeded max number of allowed failed heart beat send attempts. "
+ + "Going to stop heartbeating!");
e.printStackTrace();
- System.exit(-1);
+ throw new RuntimeException(e);
} else {
LOG.warn("Will retry heartbeat..");
}
@@ -283,7 +284,7 @@ public void run() {
}
}
- //region TonyDataFeed
+ // Start region TonyDataFeed
// TODO : currently requires caller (tf job) to provide the path to read
// maybe a better abstraction if task executor itself figures this out (if
@@ -303,30 +304,7 @@ public HdfsAvroFileSplitReader getHdfsAvroFileSplitReader(List readPaths
return new HdfsAvroFileSplitReader(hdfsConf, readPaths, this.taskIndex,
this.numTasks, useRandomShuffle);
}
-
-
- //endregion
-
- //region Testing
-
- private void hangIfTesting() {
- // Simulate hanging task executor if enabled and is first attempt
- String shouldHang = System.getenv(Constants.TEST_TASK_EXECUTOR_HANG);
- String attempt = System.getenv(Constants.ATTEMPT_NUMBER);
- int attemptNumber = attempt == null ? 0 : Integer.valueOf(attempt);
- if (shouldHang != null && Boolean.parseBoolean(shouldHang) && attemptNumber < 1) {
- LOG.info("Hanging for 20 seconds for testing purposes");
- try {
- Thread.sleep(20000);
- } catch (InterruptedException e) {
- LOG.error("Thread interrupted while hanging forever", e);
- }
- // We still exit after 20 seconds to prevent this process from sticking around forever.
- // In the cluster, when using cgroups, when the container for this process is killed, this process will also be
- // killed, but when using MiniYARNCluster, that's not the case, so this process still needs to exit during tests.
- System.exit(-1);
- }
- }
+ // End region TonyDataFeed
private void skewAndHangIfTesting() {
String skewInstr = System.getenv(Constants.TEST_TASK_EXECUTOR_SKEW);
@@ -349,6 +327,4 @@ private void skewAndHangIfTesting() {
}
}
}
- //endregion
-
}
diff --git a/tony-core/src/main/java/com/linkedin/tony/TonyApplicationMaster.java b/tony-core/src/main/java/com/linkedin/tony/TonyApplicationMaster.java
index 22568b51..5fdc2bf4 100644
--- a/tony-core/src/main/java/com/linkedin/tony/TonyApplicationMaster.java
+++ b/tony-core/src/main/java/com/linkedin/tony/TonyApplicationMaster.java
@@ -24,11 +24,13 @@
import com.linkedin.tony.util.Utils;
import java.io.BufferedReader;
import java.io.File;
-import java.io.FileReader;
+import java.io.FileInputStream;
import java.io.IOException;
+import java.io.InputStreamReader;
import java.lang.reflect.Method;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -86,8 +88,6 @@
import org.apache.hadoop.yarn.util.AbstractLivelinessMonitor;
import py4j.GatewayServer;
-import static com.linkedin.tony.TonyConfigurationKeys.*;
-
public class TonyApplicationMaster {
private static final Log LOG = LogFactory.getLog(TonyApplicationMaster.class);
@@ -146,7 +146,7 @@ public class TonyApplicationMaster {
private TonySession.Builder sessionBuilder;
/** Configuration **/
- private static Configuration yarnConf;
+ private Configuration yarnConf;
private Configuration hdfsConf;
/** Cluster spec **/
@@ -176,9 +176,6 @@ public class TonyApplicationMaster {
private volatile boolean taskHasMissesHB = false;
private Thread mainThread;
- /** Handle different machine frameworks **/
- private MLFramework framework;
-
private TonyApplicationMaster() {
hdfsConf = new Configuration();
yarnConf = new Configuration();
@@ -264,8 +261,6 @@ private boolean init(String[] args) {
TonyConfigurationKeys.DEFAULT_TASK_HEARTBEAT_INTERVAL_MS);
maxConsecutiveHBMiss = tonyConf.getInt(TonyConfigurationKeys.TASK_MAX_MISSED_HEARTBEATS,
TonyConfigurationKeys.DEFAULT_TASK_MAX_MISSED_HEARTBEATS);
- framework = MLFramework.valueOf(tonyConf.get(TonyConfigurationKeys.FRAMEWORK_NAME,
- TonyConfigurationKeys.DEFAULT_FRAMEWORK_NAME).toUpperCase());
tonyHistoryFolder = tonyConf.get(TonyConfigurationKeys.TONY_HISTORY_LOCATION,
TonyConfigurationKeys.DEFAULT_TONY_HISTORY_LOCATION);
@@ -509,7 +504,7 @@ private void writeConfigFile(FileSystem fs, Path jobDir) throws IOException {
if (jobDir == null) {
return;
}
- Path configFile = new Path(jobDir,"config.xml");
+ Path configFile = new Path(jobDir, "config.xml");
try (FSDataOutputStream out = fs.create(configFile)) {
tonyConf.writeXml(out);
} catch (IOException e) {
@@ -744,9 +739,9 @@ private int doPreprocessingJob() throws Exception {
session.setFinalStatus(FinalApplicationStatus.FAILED, "Preprocessing job failed.");
return exitCode;
}
- try (BufferedReader reader = new BufferedReader(new FileReader(
+ try (BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(
System.getProperty(YarnConfiguration.YARN_APP_CONTAINER_LOG_DIR)
- + File.separatorChar + Constants.AM_STDOUT_FILENAME))) {
+ + File.separatorChar + Constants.AM_STDOUT_FILENAME), StandardCharsets.UTF_8))) {
String line;
while ((line = reader.readLine()) != null) {
if (line.contains("Model parameters: ")) {
@@ -1042,7 +1037,7 @@ public void onContainersAllocated(List containers) {
+ ", containerId = " + container.getId()
+ ", containerNode = " + container.getNodeId().getHost() + ":" + container.getNodeId().getPort()
+ ", resourceRequest = " + container.getResource());
- new ContainerLauncher(container, containerListener).run();
+ new ContainerLauncher(container).run();
}
}
@@ -1074,11 +1069,9 @@ public void onError(Throwable throwable) {
*/
private class ContainerLauncher implements Runnable {
Container container;
- NMCallbackHandler containerListener;
- ContainerLauncher(Container container, NMCallbackHandler containerListener) {
+ ContainerLauncher(Container container) {
this.container = container;
- this.containerListener = containerListener;
}
/**
diff --git a/tony-core/src/main/java/com/linkedin/tony/TonyConfigurationKeys.java b/tony-core/src/main/java/com/linkedin/tony/TonyConfigurationKeys.java
index c8aa1894..7f6c59a3 100644
--- a/tony-core/src/main/java/com/linkedin/tony/TonyConfigurationKeys.java
+++ b/tony-core/src/main/java/com/linkedin/tony/TonyConfigurationKeys.java
@@ -18,7 +18,7 @@ private TonyConfigurationKeys() {
// Version info configuration
public static final String TONY_VERSION_INFO_PREFIX = TONY_PREFIX + "version-info.";
- public static final String TONY_VERSION_INFO_VERSION= TONY_VERSION_INFO_PREFIX + "version";
+ public static final String TONY_VERSION_INFO_VERSION = TONY_VERSION_INFO_PREFIX + "version";
public static final String TONY_VERSION_INFO_REVISION = TONY_VERSION_INFO_PREFIX + "revision";
public static final String TONY_VERSION_INFO_BRANCH = TONY_VERSION_INFO_PREFIX + "branch";
public static final String TONY_VERSION_INFO_USER = TONY_VERSION_INFO_PREFIX + "user";
diff --git a/tony-core/src/main/java/com/linkedin/tony/io/HdfsAvroFileSplitReader.java b/tony-core/src/main/java/com/linkedin/tony/io/HdfsAvroFileSplitReader.java
index 9618464e..06ce8428 100644
--- a/tony-core/src/main/java/com/linkedin/tony/io/HdfsAvroFileSplitReader.java
+++ b/tony-core/src/main/java/com/linkedin/tony/io/HdfsAvroFileSplitReader.java
@@ -415,7 +415,7 @@ public List createReadInfo(List readPaths,
return filesToRead;
}
- private class FileAccessInfo {
+ private static class FileAccessInfo {
final String filePath;
final long startOffset;
final long readLength;
@@ -472,7 +472,7 @@ public String getSchemaJson() {
* as a byte array. Python code can simply parse this
* in memory byte array as if it is a regular avro file.
*/
- class FileObject extends OutputStream {
+ static class FileObject extends OutputStream {
ByteArrayOutputStream stream;
@@ -769,7 +769,9 @@ T poll(int time, TimeUnit unit) throws InterruptedException {
int attempt = 100;
while (list.size() < pollingThreshold * this.bufferSize
&& !fetcher.readFinished && attempt-- > 0) {
- bufferReady.await(10, TimeUnit.MILLISECONDS);
+ if (!bufferReady.await(10, TimeUnit.MILLISECONDS) && attempt % 20 == 0) {
+ LOG.warn("Read buffer not ready");
+ }
}
if (attempt <= 0) {
return null;
diff --git a/tony-core/src/main/java/com/linkedin/tony/rpc/ApplicationRpcServer.java b/tony-core/src/main/java/com/linkedin/tony/rpc/ApplicationRpcServer.java
index a40c8a52..44686b93 100644
--- a/tony-core/src/main/java/com/linkedin/tony/rpc/ApplicationRpcServer.java
+++ b/tony-core/src/main/java/com/linkedin/tony/rpc/ApplicationRpcServer.java
@@ -8,6 +8,7 @@
import com.linkedin.tony.TFPolicyProvider;
import com.linkedin.tony.rpc.impl.pb.service.TensorFlowClusterPBServiceImpl;
import java.io.IOException;
+import java.util.Random;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.ipc.ProtocolSignature;
@@ -22,8 +23,8 @@
public class ApplicationRpcServer extends Thread implements TensorFlowCluster {
- private static final RecordFactory RECORD_FACTORY =
- RecordFactoryProvider.getRecordFactory(null);
+ private static final RecordFactory RECORD_FACTORY = RecordFactoryProvider.getRecordFactory(null);
+ private static final Random RANDOM_NUMBER_GENERATOR = new Random();
private final int rpcPort;
private final String rpcAddress;
private final ApplicationRpc appRpc;
@@ -33,7 +34,7 @@ public class ApplicationRpcServer extends Thread implements TensorFlowCluster {
public ApplicationRpcServer(String hostname, ApplicationRpc rpc, Configuration conf) {
this.rpcAddress = hostname;
- this.rpcPort = 10000 + ((int) (Math.random() * (5000)) + 1);
+ this.rpcPort = 10000 + RANDOM_NUMBER_GENERATOR.nextInt(5000) + 1;
this.appRpc = rpc;
this.conf = conf;
if (conf == null) {
diff --git a/tony-core/src/main/java/com/linkedin/tony/rpc/TaskUrl.java b/tony-core/src/main/java/com/linkedin/tony/rpc/TaskUrl.java
index 812f3790..0ef502ae 100644
--- a/tony-core/src/main/java/com/linkedin/tony/rpc/TaskUrl.java
+++ b/tony-core/src/main/java/com/linkedin/tony/rpc/TaskUrl.java
@@ -4,6 +4,8 @@
*/
package com.linkedin.tony.rpc;
+import java.util.Objects;
+
/**
* Contains the name, index, and URL for a task.
@@ -38,4 +40,22 @@ public int compareTo(TaskUrl other) {
}
return Integer.valueOf(this.index).compareTo(Integer.valueOf(other.index));
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TaskUrl taskUrl = (TaskUrl) o;
+ return Objects.equals(name, taskUrl.name) && Objects.equals(index, taskUrl.index) && Objects.equals(url,
+ taskUrl.url);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(name, index, url);
+ }
}
diff --git a/tony-core/src/main/java/com/linkedin/tony/tensorflow/TonySession.java b/tony-core/src/main/java/com/linkedin/tony/tensorflow/TonySession.java
index a009b8a3..821be5a9 100644
--- a/tony-core/src/main/java/com/linkedin/tony/tensorflow/TonySession.java
+++ b/tony-core/src/main/java/com/linkedin/tony/tensorflow/TonySession.java
@@ -6,7 +6,6 @@
import com.google.common.base.Preconditions;
import com.linkedin.tony.Constants;
-import com.linkedin.tony.TonyConfigurationKeys;
import com.linkedin.tony.util.Utils;
import com.linkedin.tony.rpc.TaskUrl;
import java.net.URI;
@@ -21,9 +20,7 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -100,8 +97,8 @@ private TonySession(Builder builder) {
this.jvmArgs = builder.jvmArgs;
this.tonyConf = builder.tonyConf;
- for (String jobName : containerRequests.keySet()) {
- jobTasks.put(jobName, new TonyTask[containerRequests.get(jobName).getNumInstances()]);
+ for (Map.Entry entry : containerRequests.entrySet()) {
+ jobTasks.put(entry.getKey(), new TonyTask[entry.getValue().getNumInstances()]);
}
}
@@ -122,8 +119,8 @@ public void setResources(Configuration yarnConf,
Map env = System.getenv();
String tonyConfPath = env.get(Constants.TONY_CONF_PREFIX + Constants.PATH_SUFFIX);
- long tonyConfTimestamp = Long.valueOf(env.get(Constants.TONY_CONF_PREFIX + Constants.TIMESTAMP_SUFFIX));
- long tonyConfLength = Long.valueOf(env.get(Constants.TONY_CONF_PREFIX + Constants.LENGTH_SUFFIX));
+ long tonyConfTimestamp = Long.parseLong(env.get(Constants.TONY_CONF_PREFIX + Constants.TIMESTAMP_SUFFIX));
+ long tonyConfLength = Long.parseLong(env.get(Constants.TONY_CONF_PREFIX + Constants.LENGTH_SUFFIX));
LocalResource tonyConfResource =
LocalResource.newInstance(ConverterUtils.getYarnUrlFromURI(URI.create(tonyConfPath)),
@@ -358,8 +355,8 @@ private TonyTask getTask(String jobName, String taskIndex) {
* Returns true if the job is "chief" or if there is no "chief" job and ("worker", "0") is passed in.
*/
public boolean isChief(String jobName, String index) {
- return jobName.equals(CHIEF_JOB_NAME) || (!jobTasks.containsKey(CHIEF_JOB_NAME) &&
- jobName.equals(WORKER_JOB_NAME) && index.equals("0"));
+ return jobName.equals(CHIEF_JOB_NAME) || (!jobTasks.containsKey(CHIEF_JOB_NAME)
+ && jobName.equals(WORKER_JOB_NAME) && index.equals("0"));
}
public TonyTask getTask(ContainerId containerId) {
diff --git a/tony-core/src/main/java/com/linkedin/tony/util/HdfsUtils.java b/tony-core/src/main/java/com/linkedin/tony/util/HdfsUtils.java
index 826e1dd9..6e50bcdb 100644
--- a/tony-core/src/main/java/com/linkedin/tony/util/HdfsUtils.java
+++ b/tony-core/src/main/java/com/linkedin/tony/util/HdfsUtils.java
@@ -8,6 +8,7 @@
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -75,7 +76,7 @@ public static boolean pathExists(FileSystem fs, Path filePath) {
public static String contentOfHdfsFile(FileSystem fs, Path filePath) {
StringBuilder fileContent = new StringBuilder();
try (FSDataInputStream inStrm = fs.open(filePath);
- BufferedReader bufReader = new BufferedReader(new InputStreamReader(inStrm))) {
+ BufferedReader bufReader = new BufferedReader(new InputStreamReader(inStrm, StandardCharsets.UTF_8))) {
String line;
while ((line = bufReader.readLine()) != null) {
fileContent.append(line);
diff --git a/tony-core/src/main/java/com/linkedin/tony/util/Utils.java b/tony-core/src/main/java/com/linkedin/tony/util/Utils.java
index 861a220b..ab7ace5f 100644
--- a/tony-core/src/main/java/com/linkedin/tony/util/Utils.java
+++ b/tony-core/src/main/java/com/linkedin/tony/util/Utils.java
@@ -384,7 +384,7 @@ public static String constructTFConfig(String clusterSpec, String jobName, int t
ObjectMapper mapper = new ObjectMapper();
try {
Map> spec =
- mapper.readValue(clusterSpec, new TypeReference>>() {});
+ mapper.readValue(clusterSpec, new TypeReference>>() { });
TFConfig tfConfig = new TFConfig(spec, jobName, taskIndex);
return mapper.writeValueAsString(tfConfig);
} catch (IOException ioe) {
@@ -457,13 +457,13 @@ public static void printWorkerTasksCompleted(AtomicInteger completedWTasks, long
LOG.info("Completed all " + totalWTasks + " worker tasks.");
return;
}
- LOG.info("Completed worker tasks: " + completedWTasks.get() + " out of " + totalWTasks + " worker tasks." );
+ LOG.info("Completed worker tasks: " + completedWTasks.get() + " out of " + totalWTasks + " worker tasks.");
}
public static String parseClusterSpecForPytorch(String clusterSpec) throws IOException {
ObjectMapper objectMapper = new ObjectMapper();
Map> clusterSpecMap =
- objectMapper.readValue(clusterSpec, new TypeReference>>(){});
+ objectMapper.readValue(clusterSpec, new TypeReference>>() { });
String chiefWorkerAddress = clusterSpecMap.get(Constants.WORKER_JOB_NAME).get(0);
if (chiefWorkerAddress == null) {
LOG.error("Failed to get chief worker address from cluster spec.");
diff --git a/tony-core/src/main/java/com/linkedin/tony/util/VersionInfo.java b/tony-core/src/main/java/com/linkedin/tony/util/VersionInfo.java
index 619ed4c3..287c94e9 100644
--- a/tony-core/src/main/java/com/linkedin/tony/util/VersionInfo.java
+++ b/tony-core/src/main/java/com/linkedin/tony/util/VersionInfo.java
@@ -40,42 +40,42 @@ private VersionInfo() {
}
}
- protected String _getVersion() {
+ private String getVersionInternal() {
return versionInfo.getProperty("version", "Unknown");
}
- protected String _getRevision() {
+ private String getRevisionInternal() {
return versionInfo.getProperty("revision", "Unknown");
}
- protected String _getBranch() {
+ private String getBranchInternal() {
return versionInfo.getProperty("branch", "Unknown");
}
- protected String _getDate() {
+ private String getDateInternal() {
return versionInfo.getProperty("date", "Unknown");
}
- protected String _getUser() {
+ private String getUserInternal() {
return versionInfo.getProperty("user", "Unknown");
}
- protected String _getUrl() {
+ private String getUrlInternal() {
return versionInfo.getProperty("url", "Unknown");
}
- protected String _getSrcChecksum() {
+ private String getSrcChecksumInternal() {
return versionInfo.getProperty("srcChecksum", "Unknown");
}
- private static VersionInfo COMMON_VERSION_INFO = new VersionInfo();
+ private static final VersionInfo COMMON_VERSION_INFO = new VersionInfo();
/**
* Get the jar version.
* @return the version string, e.g. "0.1.5"
*/
public static String getVersion() {
- return COMMON_VERSION_INFO._getVersion();
+ return COMMON_VERSION_INFO.getVersionInternal();
}
/**
@@ -83,7 +83,7 @@ public static String getVersion() {
* @return the commit hash, e.g. "1cba10da369b846c53a3f99e9acbf8321db876e7"
*/
public static String getRevision() {
- return COMMON_VERSION_INFO._getRevision();
+ return COMMON_VERSION_INFO.getRevisionInternal();
}
/**
@@ -91,7 +91,7 @@ public static String getRevision() {
* @return The branch name, e.g. "master" or "my-feature-branch"
*/
public static String getBranch() {
- return COMMON_VERSION_INFO._getBranch();
+ return COMMON_VERSION_INFO.getBranchInternal();
}
/**
@@ -99,7 +99,7 @@ public static String getBranch() {
* @return the compilation date in unix date format
*/
public static String getDate() {
- return COMMON_VERSION_INFO._getDate();
+ return COMMON_VERSION_INFO.getDateInternal();
}
/**
@@ -107,21 +107,21 @@ public static String getDate() {
* @return the username of the user
*/
public static String getUser() {
- return COMMON_VERSION_INFO._getUser();
+ return COMMON_VERSION_INFO.getUserInternal();
}
/**
* Get the Git URL for the root directory.
*/
public static String getUrl() {
- return COMMON_VERSION_INFO._getUrl();
+ return COMMON_VERSION_INFO.getUrlInternal();
}
/**
* Get the checksum of the source files from which this project was built.
**/
public static String getSrcChecksum() {
- return COMMON_VERSION_INFO._getSrcChecksum();
+ return COMMON_VERSION_INFO.getSrcChecksumInternal();
}
public static void injectVersionInfo(Configuration conf) {
diff --git a/tony-core/src/test/java/com/linkedin/tony/TestReader.java b/tony-core/src/test/java/com/linkedin/tony/TestReader.java
index ecf56c25..8a0bb1c3 100644
--- a/tony-core/src/test/java/com/linkedin/tony/TestReader.java
+++ b/tony-core/src/test/java/com/linkedin/tony/TestReader.java
@@ -42,20 +42,20 @@ public class TestReader {
public void testOffsetCalculation() {
for (int t = 0; t < 1000; t++) {
Random ran = new Random();
- long totalLen = Math.abs(ran.nextLong()) % 100000 + 10000;
+ long totalLen = Math.abs(ran.nextInt(Integer.MAX_VALUE)) % 100000 + 10000;
int totalIdx = ran.nextInt(20) + 10; // make sure this is not 0
- long next_start = 0;
+ long nextStart = 0;
for (int i = 0; i < totalIdx; i++) {
long start = HdfsAvroFileSplitReader.computeReadSplitStart(
totalLen, i, totalIdx);
- assertEquals(next_start, start);
+ assertEquals(nextStart, start);
long length = HdfsAvroFileSplitReader
.computeReadSplitLength(totalLen, i, totalIdx);
- next_start = start + length;
+ nextStart = start + length;
}
- assertEquals(totalLen, next_start);
+ assertEquals(totalLen, nextStart);
}
}
@@ -73,10 +73,10 @@ public void testReader() throws IOException, InterruptedException {
Files.deleteIfExists(Paths.get(path2));
try {
Schema schema = generateTestSchema();
- List all_records = new ArrayList<>();
- all_records.addAll(generateTestAvro(path0, schema));
- all_records.addAll(generateTestAvro(path1, schema));
- all_records.addAll(generateTestAvro(path2, schema));
+ List allRecords = new ArrayList<>();
+ allRecords.addAll(generateTestAvro(path0, schema));
+ allRecords.addAll(generateTestAvro(path1, schema));
+ allRecords.addAll(generateTestAvro(path2, schema));
// NOTE This will not use HDFS, this generates a RawLocalFileSystem
// instance, we will only be testing with local file system in this unit
// test.
@@ -89,8 +89,8 @@ public void testReader() throws IOException, InterruptedException {
Schema readSchema = new Schema.Parser().parse(schemaJson);
assertEquals(schema, readSchema);
- readAndCheck(reader, readSchema, all_records);
- assertEquals(all_records.size(), 0);
+ readAndCheck(reader, readSchema, allRecords);
+ assertEquals(allRecords.size(), 0);
reader.close();
} finally {
Files.deleteIfExists(Paths.get(path0));
@@ -119,10 +119,10 @@ public void testReaderPartialRead() throws IOException, InterruptedException {
List records1 = generateTestAvro(path1, schema);
List records2 = generateTestAvro(path2, schema);
- List all_records = new ArrayList<>();
- all_records.addAll(records0);
- all_records.addAll(records1);
- all_records.addAll(records2);
+ List allRecords = new ArrayList<>();
+ allRecords.addAll(records0);
+ allRecords.addAll(records1);
+ allRecords.addAll(records2);
// NOTE here we have 3 avro files, and a total of 3 splits.
// but it does not necessarily mean each reader will be processing
@@ -136,8 +136,8 @@ public void testReaderPartialRead() throws IOException, InterruptedException {
String schemaJson = reader0.getSchemaJson();
Schema readSchema = new Schema.Parser().parse(schemaJson);
assertEquals(schema, readSchema);
- // this call below will remove some entries from all_records.
- readAndCheck(reader0, readSchema, all_records);
+ // this call below will remove some entries from allRecords.
+ readAndCheck(reader0, readSchema, allRecords);
reader0.close();
HdfsAvroFileSplitReader reader1 =
@@ -148,7 +148,7 @@ public void testReaderPartialRead() throws IOException, InterruptedException {
schemaJson = reader1.getSchemaJson();
readSchema = new Schema.Parser().parse(schemaJson);
assertEquals(schema, readSchema);
- readAndCheck(reader1, readSchema, all_records);
+ readAndCheck(reader1, readSchema, allRecords);
reader1.close();
HdfsAvroFileSplitReader reader2 =
@@ -159,11 +159,11 @@ public void testReaderPartialRead() throws IOException, InterruptedException {
schemaJson = reader2.getSchemaJson();
readSchema = new Schema.Parser().parse(schemaJson);
assertEquals(schema, readSchema);
- readAndCheck(reader2, readSchema, all_records);
+ readAndCheck(reader2, readSchema, allRecords);
reader2.close();
// after all three readers, all records should be removed.
- assertEquals(all_records.size(), 0);
+ assertEquals(allRecords.size(), 0);
} finally {
Files.deleteIfExists(Paths.get(path0));
Files.deleteIfExists(Paths.get(path1));
@@ -172,7 +172,7 @@ public void testReaderPartialRead() throws IOException, InterruptedException {
}
private void readAndCheck(HdfsAvroFileSplitReader reader, Schema readSchema,
- List all_records) throws IOException, InterruptedException {
+ List allRecords) throws IOException, InterruptedException {
while (reader.hasNext()) {
// an arbitrary chosen batch size, to capture the more likely case
// that last batch will be smaller
@@ -188,7 +188,7 @@ private void readAndCheck(HdfsAvroFileSplitReader reader, Schema readSchema,
String name = record.get("name").toString();
int age = (Integer) record.get("age");
- GenericData.Record expected = all_records.remove(0);
+ GenericData.Record expected = allRecords.remove(0);
assertEquals(name, expected.get("name"));
assertEquals(age, expected.get("age"));
@@ -215,7 +215,7 @@ private List generateTestAvro(
Random ran = new Random();
File file = new File(path);
- List all_records = new ArrayList<>();
+ List allRecords = new ArrayList<>();
GenericDatumWriter datum =
new GenericDatumWriter<>(schema);
@@ -224,14 +224,14 @@ private List generateTestAvro(
writer.create(schema, file);
- for (int i = 0; i < NUM_RECORD; i ++) {
+ for (int i = 0; i < NUM_RECORD; i++) {
GenericData.Record record =
makeObject(schema, "Person " + i, ran.nextInt(120));
- all_records.add(record);
+ allRecords.add(record);
writer.append(record);
}
writer.close();
- return all_records;
+ return allRecords;
}
private static GenericData.Record makeObject(Schema schema, String name,
@@ -239,6 +239,6 @@ private static GenericData.Record makeObject(Schema schema, String name,
GenericData.Record record = new GenericData.Record(schema);
record.put("name", name);
record.put("age", age);
- return(record);
+ return record;
}
}
diff --git a/tony-core/src/test/java/com/linkedin/tony/TestTaskExecutor.java b/tony-core/src/test/java/com/linkedin/tony/TestTaskExecutor.java
index e00cc616..0d07e965 100644
--- a/tony-core/src/test/java/com/linkedin/tony/TestTaskExecutor.java
+++ b/tony-core/src/test/java/com/linkedin/tony/TestTaskExecutor.java
@@ -41,6 +41,8 @@ public void testTaskExecutorConf() throws Exception {
taskExecutor.initConfigs(args);
assertEquals(2000, taskExecutor.tonyConf.getInt(TonyConfigurationKeys.TASK_HEARTBEAT_INTERVAL_MS,
TonyConfigurationKeys.DEFAULT_TASK_HEARTBEAT_INTERVAL_MS));
- confFile.delete();
+ if (!confFile.delete()) {
+ throw new RuntimeException("Failed to delete conf file");
+ }
}
}
diff --git a/tony-core/src/test/java/com/linkedin/tony/TestTonyConfigurationFields.java b/tony-core/src/test/java/com/linkedin/tony/TestTonyConfigurationFields.java
index 5b0c73a7..5dd74833 100644
--- a/tony-core/src/test/java/com/linkedin/tony/TestTonyConfigurationFields.java
+++ b/tony-core/src/test/java/com/linkedin/tony/TestTonyConfigurationFields.java
@@ -12,7 +12,7 @@ public class TestTonyConfigurationFields extends TestConfigurationFieldsBase {
@Override
public void initializeMemberVariables() {
- xmlFilename = new String(Constants.TONY_DEFAULT_XML);
+ xmlFilename = Constants.TONY_DEFAULT_XML;
configurationClasses = new Class[] { TonyConfigurationKeys.class };
// Set error modes
diff --git a/tony-core/src/test/java/com/linkedin/tony/TestTonyE2E.java b/tony-core/src/test/java/com/linkedin/tony/TestTonyE2E.java
index d8c483be..58d14714 100644
--- a/tony-core/src/test/java/com/linkedin/tony/TestTonyE2E.java
+++ b/tony-core/src/test/java/com/linkedin/tony/TestTonyE2E.java
@@ -115,7 +115,7 @@ public void testPSSkewedWorkerTrainingShouldPass() throws ParseException, IOExce
"--python_binary_path", "python",
"--shell_env", "ENV_CHECK=ENV_CHECK",
"--container_env", Constants.SKIP_HADOOP_PATH + "=true",
- "--container_env", Constants.TEST_TASK_EXECUTOR_SKEW+ "=worker#0#30000"
+ "--container_env", Constants.TEST_TASK_EXECUTOR_SKEW + "=worker#0#30000"
});
int exitCode = client.start();
Assert.assertEquals(exitCode, 0);
@@ -205,7 +205,8 @@ public void testAMCrashTonyShouldFail() throws ParseException, IOException {
*/
@Test
public void testAMStopsJobAfterWorker0Killed() throws ParseException, IOException {
- client.init(new String[]{"--src_dir", "tony-core/src/test/resources/", "--executes", "exit_0.py", "--hdfs_classpath", "/yarn/libs", "--python_binary_path", "python", "--container_env",
+ client.init(new String[]{"--src_dir", "tony-core/src/test/resources/", "--executes", "exit_0.py",
+ "--hdfs_classpath", "/yarn/libs", "--python_binary_path", "python", "--container_env",
Constants.TEST_WORKER_TERMINATED + "=true"});
int exitCode = client.start();
Assert.assertEquals(exitCode, -1);
diff --git a/tony-core/src/test/java/com/linkedin/tony/events/TestEventHandler.java b/tony-core/src/test/java/com/linkedin/tony/events/TestEventHandler.java
index f13c92c1..c2e7572e 100644
--- a/tony-core/src/test/java/com/linkedin/tony/events/TestEventHandler.java
+++ b/tony-core/src/test/java/com/linkedin/tony/events/TestEventHandler.java
@@ -53,7 +53,7 @@ public void setup() {
}
@Test
- public void testSetUpThread_failedToSetUpWriter() throws IOException {
+ public void testSetUpThreadFailedToSetUpWriter() throws IOException {
FileSystem mockFs = mock(FileSystem.class);
when(mockFs.create(any(Path.class))).thenThrow(new IOException("IO Exception"));
@@ -65,7 +65,7 @@ public void testSetUpThread_failedToSetUpWriter() throws IOException {
}
@Test
- public void testEventHandlerE2E_success() throws InterruptedException, IOException {
+ public void testEventHandlerE2ESuccess() throws InterruptedException, IOException {
fs.mkdirs(jobDir);
eventHandlerThread = new EventHandler(fs, eventQueue);
eventHandlerThread.setUpThread(jobDir, metadata);
@@ -92,7 +92,7 @@ public void testEventHandlerE2E_success() throws InterruptedException, IOExcepti
}
@Test
- public void testEventHandlerE2E_failedJobDirNotSet() throws InterruptedException, IOException {
+ public void testEventHandlerE2EFailedJobDirNotSet() throws InterruptedException, IOException {
fs.mkdirs(jobDir);
eventHandlerThread = new EventHandler(fs, eventQueue);
eventHandlerThread.start();
diff --git a/tony-core/src/test/java/com/linkedin/tony/util/TestHdfsUtils.java b/tony-core/src/test/java/com/linkedin/tony/util/TestHdfsUtils.java
index 44457ee1..ff44c4f9 100644
--- a/tony-core/src/test/java/com/linkedin/tony/util/TestHdfsUtils.java
+++ b/tony-core/src/test/java/com/linkedin/tony/util/TestHdfsUtils.java
@@ -34,26 +34,26 @@ public static void setup() {
}
@Test
- public void testScanDir_emptyDir() {
+ public void testScanDirEmptyDir() {
Path histFolder = new Path(Constants.TONY_CORE_SRC + "test/resources/emptyHistFolder");
assertEquals(HdfsUtils.scanDir(fs, histFolder).length, 0);
}
@Test
- public void testScanDir_typical() {
+ public void testScanDirTypical() {
Path histFolder = new Path(Constants.TONY_CORE_SRC + "test/resources/typicalHistFolder");
FileStatus[] res = HdfsUtils.scanDir(fs, histFolder);
assertEquals(res.length, 5);
}
@Test
- public void testScanDir_nullDir() {
+ public void testScanDirNullDir() {
assertEquals(HdfsUtils.scanDir(fs, null).length, 0);
}
@Test
- public void testScanDir_throwsException() throws IOException {
+ public void testScanDirThrowsException() throws IOException {
FileSystem mockFs = mock(FileSystem.class);
Path invalidPath = new Path("/invalid/path");
@@ -64,21 +64,21 @@ public void testScanDir_throwsException() throws IOException {
}
@Test
- public void testPathExists_true() {
+ public void testPathExistsTrue() {
Path exists = new Path(Constants.TONY_CORE_SRC + "test/resources/file.txt");
assertTrue(HdfsUtils.pathExists(fs, exists));
}
@Test
- public void testPathExists_false() {
+ public void testPathExistsFalse() {
Path invalidPath = new Path("/invalid/path");
assertFalse(HdfsUtils.pathExists(fs, invalidPath));
}
@Test
- public void testPathExists_throwsException() throws IOException {
+ public void testPathExistsThrowsException() throws IOException {
FileSystem mockFs = mock(FileSystem.class);
Path invalidPath = new Path("/invalid/path");
@@ -88,21 +88,21 @@ public void testPathExists_throwsException() throws IOException {
}
@Test
- public void testContentOfHdfsFile_withContent() {
+ public void testContentOfHdfsFileWithContent() {
Path filePath = new Path(Constants.TONY_CORE_SRC + "test/resources/file.txt");
- assertEquals(HdfsUtils.contentOfHdfsFile(fs, filePath),"someContent");
+ assertEquals(HdfsUtils.contentOfHdfsFile(fs, filePath), "someContent");
}
@Test
- public void testContentOfHdfsFile_noContent() {
+ public void testContentOfHdfsFileNoContent() {
Path filePath = new Path(Constants.TONY_CORE_SRC + "test/resources/empty.txt");
assertEquals(HdfsUtils.contentOfHdfsFile(fs, filePath), "");
}
@Test
- public void testContentOfHdfsFile_throwsException() throws IOException {
+ public void testContentOfHdfsFileThrowsException() throws IOException {
FileSystem mockFs = mock(FileSystem.class);
Path filePath = new Path(Constants.TONY_CORE_SRC + "test/resources/empty.txt");
@@ -112,7 +112,7 @@ public void testContentOfHdfsFile_throwsException() throws IOException {
}
@Test
- public void testGetJobId_typicalCase() {
+ public void testGetJobIdTypicalCase() {
Path filePath1 = new Path(Constants.TONY_CORE_SRC + "test/resources/job1");
Path filePath2 = new Path(Constants.TONY_CORE_SRC + "test/resources/app2/");
@@ -121,7 +121,7 @@ public void testGetJobId_typicalCase() {
}
@Test
- public void testGetJobId_emptyPath() {
+ public void testGetJobIdEmptyPath() {
Path filePath = mock(Path.class);
when(filePath.toString()).thenReturn("");
@@ -129,7 +129,7 @@ public void testGetJobId_emptyPath() {
}
@Test
- public void testIsJobFolder_match() {
+ public void testIsJobFolderMatch() {
Path filePath1 = new Path("/abc/def/application_1541469337545_0024");
Path filePath2 = new Path("/abc/def/job2");
String regex1 = "^application_.*";
@@ -140,7 +140,7 @@ public void testIsJobFolder_match() {
}
@Test
- public void testIsJobFolder_notMatch() {
+ public void testIsJobFolderNoMatch() {
Path filePath1 = new Path(Constants.TONY_CORE_SRC + "test/job/application_1541469337545_0024");
Path filePath2 = new Path(Constants.TONY_CORE_SRC + "test/resources/application2/");
String regex1 = ".*job.*";
@@ -151,14 +151,14 @@ public void testIsJobFolder_notMatch() {
}
@Test
- public void testGetJobFolders_emptyHistoryFolder() {
+ public void testGetJobFoldersEmptyHistoryFolder() {
Path histFolder = new Path(Constants.TONY_CORE_SRC + "test/resources/emptyHistFolder");
assertEquals(HdfsUtils.getJobFolders(fs, histFolder, "job*"), new ArrayList());
}
@Test
- public void testGetJobFolders_typicalHistoryFolder() {
+ public void testGetJobFoldersTypicalHistoryFolder() {
Path histFolder = new Path(Constants.TONY_CORE_SRC + "test/resources/typicalHistFolder");
String regex = "^job.*";
List expectedRes = new ArrayList<>();
@@ -177,7 +177,7 @@ public void testGetJobFolders_typicalHistoryFolder() {
}
@Test
- public void testGetJobFolders_nestedHistFolder() {
+ public void testGetJobFoldersNestedHistFolder() {
Path histFolder = new Path(Constants.TONY_CORE_SRC + "test/resources/nestedHistFolder");
String regex = "^job.*";
List expectedRes = new ArrayList<>();
@@ -197,7 +197,7 @@ public void testGetJobFolders_nestedHistFolder() {
Collections.sort(actualRes, (o1, o2) -> {
String job1 = HdfsUtils.getJobId(o1.toString());
String job2 = HdfsUtils.getJobId(o2.toString());
- return job1.charAt(job1.length()-1) - job2.charAt(job2.length()-1);
+ return job1.charAt(job1.length() - 1) - job2.charAt(job2.length() - 1);
});
assertEquals(actualRes, expectedRes);
@@ -205,7 +205,7 @@ public void testGetJobFolders_nestedHistFolder() {
}
@Test
- public void testGetJobFolders_throwsException() throws IOException {
+ public void testGetJobFoldersThrowsException() throws IOException {
Path histFolder = new Path(Constants.TONY_CORE_SRC + "test/resources/typicalHistFolder");
FileSystem mockFs = mock(FileSystem.class);
String regex = "job*";
diff --git a/tony-core/src/test/java/com/linkedin/tony/util/TestHistoryFileUtils.java b/tony-core/src/test/java/com/linkedin/tony/util/TestHistoryFileUtils.java
index 3fffcb53..b96ac248 100644
--- a/tony-core/src/test/java/com/linkedin/tony/util/TestHistoryFileUtils.java
+++ b/tony-core/src/test/java/com/linkedin/tony/util/TestHistoryFileUtils.java
@@ -13,7 +13,7 @@
public class TestHistoryFileUtils {
@Test
- public void testGenerateFileName_inprogressJob() {
+ public void testGenerateFileNameInProgressJob() {
String appId = "app123";
long started = 1L;
String user = "user";
@@ -29,7 +29,7 @@ public void testGenerateFileName_inprogressJob() {
}
@Test
- public void testGenerateFileName_finishedJob() {
+ public void testGenerateFileNameFinishedJob() {
String appId = "app123";
long started = 1L;
long completed = 2L;
diff --git a/tony-core/src/test/java/com/linkedin/tony/util/TestParserUtils.java b/tony-core/src/test/java/com/linkedin/tony/util/TestParserUtils.java
index fd4da6e6..11f8bc72 100644
--- a/tony-core/src/test/java/com/linkedin/tony/util/TestParserUtils.java
+++ b/tony-core/src/test/java/com/linkedin/tony/util/TestParserUtils.java
@@ -35,7 +35,7 @@ public static void setup() {
}
@Test
- public void testIsValidHistFileName_true() {
+ public void testIsValidHistFileNameTrue() {
String fileName = "job123-1-1-user1-FAILED." + Constants.HISTFILE_SUFFIX;
String jobRegex = "job\\d+";
@@ -43,7 +43,7 @@ public void testIsValidHistFileName_true() {
}
@Test
- public void testIsValidHistFileName_false() {
+ public void testIsValidHistFileNameFalse() {
// Job name doesn't match job regex
String fileName1 = "application123-1-1-user1-FAILED." + Constants.HISTFILE_SUFFIX;
// User isn't supposed to be upper-cased
@@ -55,7 +55,7 @@ public void testIsValidHistFileName_false() {
}
@Test
- public void testParseMetadata_success() {
+ public void testParseMetadataSuccess() {
Path jobFolder = new Path(Constants.TONY_CORE_SRC + "test/resources/typicalHistFolder/job1");
String jobRegex = "application\\d+";
JobMetadata expected = new JobMetadata.Builder()
@@ -79,7 +79,7 @@ public void testParseMetadata_success() {
}
@Test
- public void testParseMetadata_fail_IOException() throws IOException {
+ public void testParseMetadataFailIOException() throws IOException {
Path jobFolder = new Path(Constants.TONY_CORE_SRC + "test/resources/typicalHistFolder/job1");
String jobRegex = "application\\d+";
FileSystem mockFs = mock(FileSystem.class);
@@ -90,7 +90,7 @@ public void testParseMetadata_fail_IOException() throws IOException {
}
@Test
- public void testParseConfig_success() {
+ public void testParseConfigSuccess() {
Path jobFolder = new Path(Constants.TONY_CORE_SRC + "test/resources/typicalHistFolder/job1");
List expected = new ArrayList<>();
JobConfig expectedConfig = new JobConfig();
@@ -110,7 +110,7 @@ public void testParseConfig_success() {
}
@Test
- public void testParseConfig_fail_IOException() throws IOException {
+ public void testParseConfigFailIOException() throws IOException {
Path jobFolder = new Path(Constants.TONY_CORE_SRC + "test/resources/typicalHistFolder/job1");
FileSystem mockFs = mock(FileSystem.class);
when(mockFs.listStatus(jobFolder)).thenThrow(new IOException("IO Excpt"));
diff --git a/tony-core/src/test/java/com/linkedin/tony/util/TestUtils.java b/tony-core/src/test/java/com/linkedin/tony/util/TestUtils.java
index 66edca16..f07b250d 100644
--- a/tony-core/src/test/java/com/linkedin/tony/util/TestUtils.java
+++ b/tony-core/src/test/java/com/linkedin/tony/util/TestUtils.java
@@ -111,7 +111,7 @@ public void testConstructTFConfig() throws IOException {
String spec = "{\"worker\":[\"host0:1234\", \"host1:1234\"], \"ps\":[\"host2:1234\"]}";
String tfConfig = Utils.constructTFConfig(spec, "worker", 1);
ObjectMapper mapper = new ObjectMapper();
- TFConfig config = mapper.readValue(tfConfig, new TypeReference(){});
+ TFConfig config = mapper.readValue(tfConfig, new TypeReference() { });
assertEquals("worker", config.getTask().getType());
assertEquals(1, config.getTask().getIndex());
assertEquals("host0:1234", config.getCluster().get("worker").get(0));
diff --git a/tony-proxy/src/main/java/com/linkedin/tonyproxy/ProxyServer.java b/tony-proxy/src/main/java/com/linkedin/tonyproxy/ProxyServer.java
index 66575367..c2c25423 100644
--- a/tony-proxy/src/main/java/com/linkedin/tonyproxy/ProxyServer.java
+++ b/tony-proxy/src/main/java/com/linkedin/tonyproxy/ProxyServer.java
@@ -8,8 +8,6 @@
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
-import java.io.OutputStreamWriter;
-import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
import org.apache.commons.logging.Log;
@@ -61,10 +59,10 @@ public void run() {
final byte[] request = new byte[1024];
byte[] reply = new byte[4096];
new Thread(() -> {
- int bytes_read;
+ int bytesRead;
try {
- while ((bytes_read = inFromClient.read(request)) != -1) {
- outToServer.write(request, 0, bytes_read);
+ while ((bytesRead = inFromClient.read(request)) != -1) {
+ outToServer.write(request, 0, bytesRead);
outToServer.flush();
}
outToServer.close();
@@ -72,9 +70,9 @@ public void run() {
LOG.error(e);
}
}).start();
- int bytes_read;
- while ((bytes_read = inFromServer.read(reply)) != -1) {
- outToClient.write(reply, 0, bytes_read);
+ int bytesRead;
+ while ((bytesRead = inFromServer.read(reply)) != -1) {
+ outToClient.write(reply, 0, bytesRead);
outToClient.flush();
}
} catch (IOException e) {