You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@apex.apache.org by th...@apache.org on 2016/04/09 05:52:34 UTC
[16/16] incubator-apex-core git commit: APEXCORE-423 style fix for
apex-core engine
APEXCORE-423 style fix for apex-core engine
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/85db448e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/85db448e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/85db448e
Branch: refs/heads/master
Commit: 85db448ef11ca3573f721e1f1235d967e0e9da66
Parents: 2af925c
Author: CI Support <je...@datatorrent.com>
Authored: Wed Apr 6 15:02:26 2016 -0700
Committer: David Yan <da...@datatorrent.com>
Committed: Fri Apr 8 17:57:05 2016 -0700
----------------------------------------------------------------------
.idea/codeStyleSettings.xml | 2 +-
.../java/com/datatorrent/api/Attribute.java | 8 +-
.../java/com/datatorrent/api/StringCodec.java | 20 +-
.../bufferserver/internal/FastDataList.java | 3 +-
.../datatorrent/bufferserver/server/Server.java | 6 +-
.../common/util/AsyncFSStorageAgent.java | 4 +-
.../datatorrent/common/util/FSStorageAgent.java | 11 +-
.../common/util/PubSubWebSocketClient.java | 4 +-
engine/engine-checkstyle-suppressions.xml | 4 +
engine/pom.xml | 3 +-
.../BlacklistBasedResourceRequestHandler.java | 6 +-
.../stram/BufferServerController.java | 2 +-
.../com/datatorrent/stram/FSEventRecorder.java | 40 +-
.../datatorrent/stram/FSRecoveryHandler.java | 91 +-
.../com/datatorrent/stram/FSStatsRecorder.java | 67 +-
.../java/com/datatorrent/stram/Journal.java | 40 +-
.../stram/LaunchContainerRunnable.java | 61 +-
.../com/datatorrent/stram/LocalModeImpl.java | 10 +-
.../datatorrent/stram/RecoverableRpcProxy.java | 50 +-
.../stram/ResourceRequestHandler.java | 13 +-
.../com/datatorrent/stram/StramAppContext.java | 10 +-
.../java/com/datatorrent/stram/StramClient.java | 101 +--
.../datatorrent/stram/StramLocalCluster.java | 36 +-
.../java/com/datatorrent/stram/StramUtils.java | 13 +-
.../datatorrent/stram/StreamingAppMaster.java | 5 +-
.../stram/StreamingAppMasterService.java | 45 +-
.../stram/StreamingContainerAgent.java | 87 +-
.../stram/StreamingContainerManager.java | 367 ++++----
.../stram/StreamingContainerParent.java | 91 +-
.../stram/StreamingContainerSecurityInfo.java | 8 +-
.../com/datatorrent/stram/StringCodecs.java | 17 +-
.../datatorrent/stram/api/ContainerContext.java | 8 +-
.../datatorrent/stram/api/ContainerEvent.java | 20 +-
.../stram/api/OperatorDeployInfo.java | 40 +-
.../datatorrent/stram/api/RequestFactory.java | 5 +-
.../com/datatorrent/stram/api/StramEvent.java | 3 +-
.../api/StramToNodeGetPropertyRequest.java | 11 +-
.../StreamingContainerUmbilicalProtocol.java | 107 ++-
.../stram/appdata/AppDataPushAgent.java | 32 +-
.../java/com/datatorrent/stram/cli/DTCli.java | 862 +++++++++----------
.../datatorrent/stram/client/AppPackage.java | 57 +-
.../stram/client/ClassPathResolvers.java | 41 +-
.../datatorrent/stram/client/ConfigPackage.java | 36 +-
.../stram/client/DTConfiguration.java | 71 +-
.../datatorrent/stram/client/EventsAgent.java | 88 +-
.../com/datatorrent/stram/client/FSAgent.java | 22 +-
.../stram/client/FSPartFileAgent.java | 17 +-
.../stram/client/PermissionsInfo.java | 3 +-
.../stram/client/RecordingsAgent.java | 210 +++--
.../datatorrent/stram/client/StatsAgent.java | 92 +-
.../datatorrent/stram/client/StramAgent.java | 61 +-
.../stram/client/StramAppLauncher.java | 105 +--
.../stram/client/StramClientUtils.java | 179 ++--
.../client/WebServicesVersionConversion.java | 23 +-
.../stram/codec/DefaultStatefulStreamCodec.java | 25 +-
.../stram/codec/LogicalPlanSerializer.java | 95 +-
.../com/datatorrent/stram/debug/MuxSink.java | 7 +-
.../stram/debug/OperatorIdPortNamePair.java | 3 +-
.../datatorrent/stram/debug/StdOutErrLog.java | 3 +-
.../stram/debug/TappedReservoir.java | 5 +-
.../datatorrent/stram/debug/TupleRecorder.java | 60 +-
.../stram/debug/TupleRecorderCollection.java | 62 +-
.../engine/BufferServerStatsSubscriber.java | 14 +-
.../stram/engine/DefaultUnifier.java | 2 +-
.../datatorrent/stram/engine/GenericNode.java | 77 +-
.../com/datatorrent/stram/engine/InputNode.java | 42 +-
.../datatorrent/stram/engine/MuxReservoir.java | 15 +-
.../java/com/datatorrent/stram/engine/Node.java | 107 +--
.../com/datatorrent/stram/engine/OiONode.java | 3 +-
.../datatorrent/stram/engine/PortContext.java | 2 +-
.../com/datatorrent/stram/engine/Slider.java | 66 +-
.../datatorrent/stram/engine/StreamContext.java | 8 +-
.../stram/engine/StreamingContainer.java | 90 +-
.../datatorrent/stram/engine/UnifierNode.java | 3 +-
.../stram/engine/WindowGenerator.java | 48 +-
.../engine/WindowIdActivatedReservoir.java | 6 +-
.../plan/logical/DefaultKryoStreamCodec.java | 10 +-
.../plan/logical/LogicalOperatorStatus.java | 1 +
.../stram/plan/logical/LogicalPlan.java | 205 ++---
.../plan/logical/LogicalPlanConfiguration.java | 383 ++++----
.../stram/plan/logical/Operators.java | 23 +-
.../StreamCodecWrapperForPersistance.java | 4 +-
.../logical/requests/CreateOperatorRequest.java | 2 +-
.../logical/requests/LogicalPlanRequest.java | 9 +-
.../stram/plan/physical/OperatorStatus.java | 6 +-
.../stram/plan/physical/PTContainer.java | 86 +-
.../stram/plan/physical/PTOperator.java | 109 ++-
.../stram/plan/physical/PhysicalPlan.java | 255 +++---
.../stram/plan/physical/PlanModifier.java | 24 +-
.../stram/plan/physical/StatsRevisions.java | 4 +-
.../stram/plan/physical/StreamMapping.java | 51 +-
.../stram/security/KerberosAuth.java | 33 +-
.../StramDelegationTokenIdentifier.java | 6 +-
.../security/StramDelegationTokenManager.java | 8 +-
.../security/StramDelegationTokenSelector.java | 13 +-
.../stram/security/StramUserLogin.java | 35 +-
.../stram/security/StramWSFilter.java | 30 +-
.../security/StramWSFilterInitializer.java | 4 +-
.../stram/security/StramWSPrincipal.java | 9 +-
.../security/StramWSServletRequestWrapper.java | 18 +-
.../stram/stream/BufferServerPublisher.java | 30 +-
.../stram/stream/BufferServerSubscriber.java | 36 +-
.../datatorrent/stram/stream/FastPublisher.java | 143 ++-
.../datatorrent/stram/stream/InlineStream.java | 3 +-
.../com/datatorrent/stram/stream/MuxStream.java | 8 +-
.../com/datatorrent/stram/stream/OiOStream.java | 9 +-
.../stram/stream/PartitionAwareSink.java | 12 +-
.../PartitionAwareSinkForPersistence.java | 5 +-
.../stram/util/AbstractWritableAdapter.java | 18 +-
.../com/datatorrent/stram/util/ConfigUtils.java | 24 +-
.../datatorrent/stram/util/FSJsonLineFile.java | 9 +-
.../stram/util/FSPartFileCollection.java | 24 +-
.../java/com/datatorrent/stram/util/FSUtil.java | 66 +-
.../stram/util/HeaderClientFilter.java | 20 +-
.../datatorrent/stram/util/MovingAverage.java | 51 +-
.../stram/util/ObjectMapperFactory.java | 20 +-
.../stram/util/PubSubWebSocketServlet.java | 62 +-
.../datatorrent/stram/util/SecureExecutor.java | 23 +-
.../stram/util/SharedPubSubWebSocketClient.java | 30 +-
.../stram/util/StablePriorityQueue.java | 77 +-
.../datatorrent/stram/util/StableWrapper.java | 71 +-
.../datatorrent/stram/util/StreamGobbler.java | 10 +-
.../com/datatorrent/stram/util/VersionInfo.java | 30 +-
.../stram/util/WebServicesClient.java | 47 +-
.../com/datatorrent/stram/webapp/AppInfo.java | 73 +-
.../datatorrent/stram/webapp/ContainerInfo.java | 3 +-
.../stram/webapp/ContainersInfo.java | 2 +-
.../stram/webapp/LogicalOperatorInfo.java | 9 +-
.../stram/webapp/LogicalOperatorsInfo.java | 14 +-
.../stram/webapp/OperatorAggregationInfo.java | 3 +-
.../stram/webapp/OperatorDiscoverer.java | 227 ++---
.../datatorrent/stram/webapp/OperatorInfo.java | 18 +-
.../datatorrent/stram/webapp/OperatorsInfo.java | 14 +-
.../datatorrent/stram/webapp/StramWebApp.java | 33 +-
.../stram/webapp/StramWebServices.java | 140 +--
.../datatorrent/stram/webapp/StreamInfo.java | 6 +-
.../datatorrent/stram/webapp/StreamsInfo.java | 13 +-
.../stram/webapp/TypeDiscoverer.java | 40 +-
.../com/datatorrent/stram/webapp/TypeGraph.java | 125 ++-
.../datatorrent/stram/webapp/asm/ASMUtil.java | 81 +-
.../stram/webapp/asm/BaseSignatureVisitor.java | 50 +-
.../stram/webapp/asm/ClassNodeType.java | 17 +-
.../stram/webapp/asm/ClassSignatureVisitor.java | 42 +-
.../stram/webapp/asm/CompactAnnotationNode.java | 20 +-
.../stram/webapp/asm/CompactClassNode.java | 22 +-
.../stram/webapp/asm/CompactFieldNode.java | 40 +-
.../stram/webapp/asm/CompactUtil.java | 27 +-
.../stram/webapp/asm/FieldSignatureVisitor.java | 21 +-
.../stram/webapp/asm/MethodNode.java | 2 -
.../webapp/asm/MethodSignatureVisitor.java | 27 +-
.../com/datatorrent/stram/webapp/asm/Type.java | 107 ++-
.../datatorrent/stram/AffinityRulesTest.java | 5 +-
.../stram/GenericOperatorPropertyCodecTest.java | 5 +-
.../com/datatorrent/stram/HostLocalTest.java | 49 +-
.../java/com/datatorrent/stram/InlineAM.java | 12 +-
.../com/datatorrent/stram/LocalityTest.java | 22 +-
.../stram/LogicalPlanModificationTest.java | 15 +-
.../com/datatorrent/stram/MockContainer.java | 25 +-
.../datatorrent/stram/OutputUnifiedTest.java | 6 +-
.../com/datatorrent/stram/PartitioningTest.java | 35 +-
.../stram/StramLocalClusterTest.java | 13 +-
.../datatorrent/stram/StramMiniClusterTest.java | 100 ++-
.../datatorrent/stram/StramRecoveryTest.java | 56 +-
.../com/datatorrent/stram/StreamCodecTest.java | 76 +-
.../stram/StreamingContainerManagerTest.java | 98 ++-
.../com/datatorrent/stram/cli/DTCliTest.java | 18 +-
.../stram/client/AppPackageTest.java | 22 +-
.../stram/client/ConfigPackageTest.java | 17 +-
.../WebServicesVersionConversionTest.java | 19 +-
.../codec/DefaultStatefulStreamCodecTest.java | 12 +-
.../datatorrent/stram/debug/MuxSinkTest.java | 1 +
.../stram/debug/TupleRecorderTest.java | 14 +-
.../stram/engine/AtMostOnceTest.java | 7 +-
.../stram/engine/AutoMetricTest.java | 17 +-
.../stram/engine/GenericNodeTest.java | 16 +-
.../stram/engine/GenericOperatorProperty.java | 3 +-
.../stram/engine/GenericTestOperator.java | 69 +-
.../datatorrent/stram/engine/InputNodeTest.java | 11 +-
.../stram/engine/InputOperatorTest.java | 44 +-
.../com/datatorrent/stram/engine/NodeTest.java | 6 +-
.../stram/engine/ProcessingModeTests.java | 20 +-
.../stram/engine/RecoverableInputOperator.java | 12 +-
.../datatorrent/stram/engine/SliderTest.java | 11 +-
.../com/datatorrent/stram/engine/StatsTest.java | 19 +-
.../stram/engine/StreamingContainerTest.java | 8 +-
.../stram/engine/TestAppDataQueryOperator.java | 4 +-
.../stram/engine/TestAppDataResultOperator.java | 2 +-
.../stram/engine/TestAppDataSourceOperator.java | 8 +-
.../engine/TestGeneratorInputOperator.java | 26 +-
.../TestNonOptionalOutportInputOperator.java | 5 +-
.../stram/engine/TestOutputOperator.java | 41 +-
.../com/datatorrent/stram/engine/TestSink.java | 5 +-
.../stram/engine/WindowGeneratorTest.java | 23 +-
.../stram/moduleexperiment/InjectConfig.java | 6 +-
.../moduleexperiment/InjectConfigTest.java | 181 ++--
.../stram/plan/SchemaTestOperator.java | 4 +-
.../stram/plan/StreamPersistanceTests.java | 154 ++--
.../datatorrent/stram/plan/TestPlanContext.java | 14 +-
.../logical/LogicalPlanConfigurationTest.java | 439 +++++-----
.../stram/plan/logical/LogicalPlanTest.java | 254 +++---
.../stram/plan/logical/MockStorageAgent.java | 5 +-
.../stram/plan/physical/PhysicalPlanTest.java | 278 +++---
.../stream/BufferServerSubscriberTest.java | 8 +-
.../stram/stream/FastPublisherTest.java | 22 +-
.../stram/stream/FastStreamTest.java | 10 +-
.../stram/stream/InlineStreamTest.java | 16 +-
.../stram/stream/OiOEndWindowTest.java | 17 +-
.../datatorrent/stram/stream/OiOStreamTest.java | 72 +-
.../stram/stream/SocketStreamTest.java | 32 +-
.../support/ManualScheduledExecutorService.java | 1 +
.../stram/support/StramTestSupport.java | 8 +-
.../stram/util/SecurityUtilsTest.java | 6 +-
.../stram/util/StablePriorityQueueTest.java | 11 +-
.../stram/webapp/OperatorDiscoveryTest.java | 170 ++--
.../stram/webapp/StramWebServicesTest.java | 94 +-
.../stram/webapp/TypeDiscoveryTest.java | 60 +-
.../webapp/asm/FastClassIndexReaderTest.java | 7 +-
217 files changed, 5288 insertions(+), 4951 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/.idea/codeStyleSettings.xml
----------------------------------------------------------------------
diff --git a/.idea/codeStyleSettings.xml b/.idea/codeStyleSettings.xml
index 4ac0d16..c199681 100644
--- a/.idea/codeStyleSettings.xml
+++ b/.idea/codeStyleSettings.xml
@@ -106,4 +106,4 @@
</option>
<option name="USE_PER_PROJECT_SETTINGS" value="true" />
</component>
-</project>
\ No newline at end of file
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/api/src/main/java/com/datatorrent/api/Attribute.java
----------------------------------------------------------------------
diff --git a/api/src/main/java/com/datatorrent/api/Attribute.java b/api/src/main/java/com/datatorrent/api/Attribute.java
index fe6907a..8dede2f 100644
--- a/api/src/main/java/com/datatorrent/api/Attribute.java
+++ b/api/src/main/java/com/datatorrent/api/Attribute.java
@@ -27,14 +27,14 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
+import com.google.common.base.Throwables;
+
import com.datatorrent.api.StringCodec.Boolean2String;
import com.datatorrent.api.StringCodec.Enum2String;
import com.datatorrent.api.StringCodec.Integer2String;
import com.datatorrent.api.StringCodec.Long2String;
import com.datatorrent.api.StringCodec.String2String;
-import com.datatorrent.netlet.util.DTThrowable;
-
/**
* Attribute represents the attribute which can be set on various components in the system.
*
@@ -256,7 +256,7 @@ public class Attribute<T> implements Serializable
}
}
} catch (Exception ex) {
- DTThrowable.rethrow(ex);
+ throw Throwables.propagate(ex);
}
return result;
}
@@ -318,7 +318,7 @@ public class Attribute<T> implements Serializable
}
}
} catch (Exception ex) {
- DTThrowable.rethrow(ex);
+ throw Throwables.propagate(ex);
}
map.put(clazz, set);
return (long)clazz.getModifiers() << 32 | clazz.hashCode();
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/api/src/main/java/com/datatorrent/api/StringCodec.java
----------------------------------------------------------------------
diff --git a/api/src/main/java/com/datatorrent/api/StringCodec.java b/api/src/main/java/com/datatorrent/api/StringCodec.java
index b92f609..2cd6a4f 100644
--- a/api/src/main/java/com/datatorrent/api/StringCodec.java
+++ b/api/src/main/java/com/datatorrent/api/StringCodec.java
@@ -62,7 +62,7 @@ public interface StringCodec<T>
*/
String toString(T pojo);
- public class String2String implements StringCodec<String>, Serializable
+ class String2String implements StringCodec<String>, Serializable
{
@Override
public String fromString(String string)
@@ -79,7 +79,7 @@ public interface StringCodec<T>
private static final long serialVersionUID = 201310141156L;
}
- public class Integer2String implements StringCodec<Integer>, Serializable
+ class Integer2String implements StringCodec<Integer>, Serializable
{
@Override
public Integer fromString(String string)
@@ -96,7 +96,7 @@ public interface StringCodec<T>
private static final long serialVersionUID = 201310141157L;
}
- public class Long2String implements StringCodec<Long>, Serializable
+ class Long2String implements StringCodec<Long>, Serializable
{
@Override
public Long fromString(String string)
@@ -113,7 +113,7 @@ public interface StringCodec<T>
private static final long serialVersionUID = 201310141158L;
}
- public class Boolean2String implements StringCodec<Boolean>, Serializable
+ class Boolean2String implements StringCodec<Boolean>, Serializable
{
@Override
public Boolean fromString(String string)
@@ -146,7 +146,7 @@ public interface StringCodec<T>
*
* @param <T> Type of the object which is converted to/from String
*/
- public class Object2String<T> implements StringCodec<T>, Serializable
+ class Object2String<T> implements StringCodec<T>, Serializable
{
public final String separator;
public final String propertySeparator;
@@ -217,7 +217,7 @@ public interface StringCodec<T>
private static final long serialVersionUID = 201311141853L;
}
- public class Map2String<K, V> implements StringCodec<Map<K, V>>, Serializable
+ class Map2String<K, V> implements StringCodec<Map<K, V>>, Serializable
{
private final StringCodec<K> keyCodec;
private final StringCodec<V> valueCodec;
@@ -274,7 +274,7 @@ public interface StringCodec<T>
private static final long serialVersionUID = 201402272053L;
}
- public class Collection2String<T> implements StringCodec<Collection<T>>, Serializable
+ class Collection2String<T> implements StringCodec<Collection<T>>, Serializable
{
private final String separator;
private final StringCodec<T> codec;
@@ -330,7 +330,7 @@ public interface StringCodec<T>
private static final long serialVersionUID = 201401091806L;
}
- public class Enum2String<T extends Enum<T>> implements StringCodec<T>, Serializable
+ class Enum2String<T extends Enum<T>> implements StringCodec<T>, Serializable
{
private final Class<T> clazz;
@@ -354,7 +354,7 @@ public interface StringCodec<T>
private static final long serialVersionUID = 201310181757L;
}
- public class Class2String<T> implements StringCodec<Class<? extends T>>, Serializable
+ class Class2String<T> implements StringCodec<Class<? extends T>>, Serializable
{
@Override
@SuppressWarnings({"BroadCatchBlock", "TooBroadCatch"})
@@ -378,7 +378,7 @@ public interface StringCodec<T>
private static final long serialVersionUID = 201312082053L;
}
- public class JsonStringCodec<T> implements StringCodec<T>, Serializable
+ class JsonStringCodec<T> implements StringCodec<T>, Serializable
{
private static final long serialVersionUID = 2513932518264776006L;
Class<?> clazz;
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/FastDataList.java
----------------------------------------------------------------------
diff --git a/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/FastDataList.java b/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/FastDataList.java
index af47f23..9baf111 100644
--- a/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/FastDataList.java
+++ b/bufferserver/src/main/java/com/datatorrent/bufferserver/internal/FastDataList.java
@@ -94,8 +94,7 @@ public class FastDataList extends DataList
}
break;
}
- }
- while (true);
+ } while (true);
last.writingOffset = writeOffset;
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/bufferserver/src/main/java/com/datatorrent/bufferserver/server/Server.java
----------------------------------------------------------------------
diff --git a/bufferserver/src/main/java/com/datatorrent/bufferserver/server/Server.java b/bufferserver/src/main/java/com/datatorrent/bufferserver/server/Server.java
index 85030dd..83b50d2 100644
--- a/bufferserver/src/main/java/com/datatorrent/bufferserver/server/Server.java
+++ b/bufferserver/src/main/java/com/datatorrent/bufferserver/server/Server.java
@@ -722,8 +722,7 @@ public class Server implements ServerListener
}
return true;
}
- }
- while (true);
+ } while (true);
}
private boolean switchToNewBufferOrSuspendRead(final byte[] array, final int offset, final int size)
@@ -862,8 +861,7 @@ public class Server implements ServerListener
read(remainingCapacity);
offset += remainingCapacity;
- }
- while ((len -= remainingCapacity) > 0);
+ } while ((len -= remainingCapacity) > 0);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/common/src/main/java/com/datatorrent/common/util/AsyncFSStorageAgent.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/com/datatorrent/common/util/AsyncFSStorageAgent.java b/common/src/main/java/com/datatorrent/common/util/AsyncFSStorageAgent.java
index e28c0e0..01a23d8 100644
--- a/common/src/main/java/com/datatorrent/common/util/AsyncFSStorageAgent.java
+++ b/common/src/main/java/com/datatorrent/common/util/AsyncFSStorageAgent.java
@@ -38,7 +38,7 @@ import org.apache.hadoop.fs.Options;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
-import com.datatorrent.netlet.util.DTThrowable;
+import com.google.common.base.Throwables;
/**
* <p>AsyncFSStorageAgent class.</p>
@@ -123,7 +123,7 @@ public class AsyncFSStorageAgent extends FSStorageAgent
} catch (Throwable t) {
logger.debug("while saving {} {}", operatorId, window, t);
stateSaved = false;
- DTThrowable.rethrow(t);
+ throw Throwables.propagate(t);
} finally {
try {
if (stream != null) {
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/common/src/main/java/com/datatorrent/common/util/FSStorageAgent.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/com/datatorrent/common/util/FSStorageAgent.java b/common/src/main/java/com/datatorrent/common/util/FSStorageAgent.java
index fd4c450..fe90b86 100644
--- a/common/src/main/java/com/datatorrent/common/util/FSStorageAgent.java
+++ b/common/src/main/java/com/datatorrent/common/util/FSStorageAgent.java
@@ -24,6 +24,7 @@ import java.io.ObjectStreamException;
import java.io.OutputStream;
import java.io.Serializable;
import java.net.URI;
+import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
@@ -43,11 +44,10 @@ import org.apache.hadoop.fs.RemoteIterator;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
-import com.google.common.collect.Lists;
+import com.google.common.base.Throwables;
import com.datatorrent.api.StorageAgent;
import com.datatorrent.api.annotation.Stateless;
-import com.datatorrent.netlet.util.DTThrowable;
/**
* FSStorageAgent
@@ -107,7 +107,7 @@ public class FSStorageAgent implements StorageAgent, Serializable
} catch (Throwable t) {
logger.debug("while saving {} {}", operatorId, window, t);
stateSaved = false;
- DTThrowable.rethrow(t);
+ throw Throwables.propagate(t);
} finally {
try {
if (stream != null) {
@@ -158,7 +158,7 @@ public class FSStorageAgent implements StorageAgent, Serializable
if (!fileStatusRemoteIterator.hasNext()) {
throw new IOException("Storage Agent has not saved anything yet!");
}
- List<Long> lwindows = Lists.newArrayList();
+ List<Long> lwindows = new ArrayList<>();
do {
FileStatus fileStatus = fileStatusRemoteIterator.next();
String name = fileStatus.getPath().getName();
@@ -166,8 +166,7 @@ public class FSStorageAgent implements StorageAgent, Serializable
continue;
}
lwindows.add(STATELESS_CHECKPOINT_WINDOW_ID.equals(name) ? Stateless.WINDOW_ID : Long.parseLong(name, 16));
- }
- while (fileStatusRemoteIterator.hasNext());
+ } while (fileStatusRemoteIterator.hasNext());
long[] windowIds = new long[lwindows.size()];
for (int i = 0; i < windowIds.length; i++) {
windowIds[i] = lwindows.get(i);
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/common/src/main/java/com/datatorrent/common/util/PubSubWebSocketClient.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/com/datatorrent/common/util/PubSubWebSocketClient.java b/common/src/main/java/com/datatorrent/common/util/PubSubWebSocketClient.java
index c3f5961..70c8570 100644
--- a/common/src/main/java/com/datatorrent/common/util/PubSubWebSocketClient.java
+++ b/common/src/main/java/com/datatorrent/common/util/PubSubWebSocketClient.java
@@ -35,6 +35,7 @@ import org.codehaus.jettison.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.base.Throwables;
import com.ning.http.client.AsyncCompletionHandler;
import com.ning.http.client.AsyncHttpClient;
import com.ning.http.client.AsyncHttpClient.BoundRequestBuilder;
@@ -48,7 +49,6 @@ import com.ning.http.client.websocket.WebSocketUpgradeHandler;
import com.datatorrent.api.Component;
import com.datatorrent.api.Context;
import com.datatorrent.common.util.PubSubMessage.PubSubMessageType;
-import com.datatorrent.netlet.util.DTThrowable;
/**
* <p>Abstract PubSubWebSocketClient class.</p>
@@ -292,7 +292,7 @@ public abstract class PubSubWebSocketClient implements Component<Context>
if (t instanceof IOException) {
throw (IOException)t;
} else {
- DTThrowable.rethrow(t);
+ throw Throwables.propagate(t);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/engine-checkstyle-suppressions.xml
----------------------------------------------------------------------
diff --git a/engine/engine-checkstyle-suppressions.xml b/engine/engine-checkstyle-suppressions.xml
index 7496011..9614eef 100644
--- a/engine/engine-checkstyle-suppressions.xml
+++ b/engine/engine-checkstyle-suppressions.xml
@@ -25,4 +25,8 @@
<suppressions>
<suppress checks="RegexpMultiline" files="DTCli.java"/>
+ <suppress checks="RegexpMultiline" files=".*Tests?.java"/>
+ <suppress checks="RegexpMultiline" files="VersionInfo.java"/>
+ <suppress checks="NoFinalizer" files="FSRecoveryHandler.java"/>
+ <suppress checks="AnnotationLocation" files=".*.java"/>
</suppressions>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/pom.xml
----------------------------------------------------------------------
diff --git a/engine/pom.xml b/engine/pom.xml
index 954b6de..490cbe3 100644
--- a/engine/pom.xml
+++ b/engine/pom.xml
@@ -34,7 +34,7 @@
<name>Apache Apex (incubating) Stream Processing Engine</name>
<properties>
- <checkstyle.console>false</checkstyle.console>
+ <checkstyle.console>true</checkstyle.console>
</properties>
<build>
@@ -149,7 +149,6 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<configuration>
- <maxAllowedViolations>3161</maxAllowedViolations>
<logViolationsToConsole>${checkstyle.console}</logViolationsToConsole>
<suppressionsLocation>engine-checkstyle-suppressions.xml</suppressionsLocation>
</configuration>
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/BlacklistBasedResourceRequestHandler.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/BlacklistBasedResourceRequestHandler.java b/engine/src/main/java/com/datatorrent/stram/BlacklistBasedResourceRequestHandler.java
index 511c66d..510f29a 100644
--- a/engine/src/main/java/com/datatorrent/stram/BlacklistBasedResourceRequestHandler.java
+++ b/engine/src/main/java/com/datatorrent/stram/BlacklistBasedResourceRequestHandler.java
@@ -24,14 +24,14 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.tuple.MutablePair;
import org.apache.hadoop.yarn.client.api.AMRMClient;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import com.datatorrent.stram.StreamingContainerAgent.ContainerStartRequest;
/**
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/BufferServerController.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/BufferServerController.java b/engine/src/main/java/com/datatorrent/stram/BufferServerController.java
index d1db29e..fb55cca 100644
--- a/engine/src/main/java/com/datatorrent/stram/BufferServerController.java
+++ b/engine/src/main/java/com/datatorrent/stram/BufferServerController.java
@@ -18,13 +18,13 @@
*/
package com.datatorrent.stram;
-import com.datatorrent.stram.engine.StreamingContainer;
import java.net.InetSocketAddress;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.datatorrent.bufferserver.client.Controller;
+import com.datatorrent.stram.engine.StreamingContainer;
/**
* Encapsulates buffer server control interface, used by the master for purging data.
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/FSEventRecorder.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/FSEventRecorder.java b/engine/src/main/java/com/datatorrent/stram/FSEventRecorder.java
index f8aac29..dfd982a 100644
--- a/engine/src/main/java/com/datatorrent/stram/FSEventRecorder.java
+++ b/engine/src/main/java/com/datatorrent/stram/FSEventRecorder.java
@@ -22,25 +22,28 @@ import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.URI;
import java.util.Map;
-import java.util.concurrent.*;
-import net.engio.mbassy.listener.Handler;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.commons.beanutils.BeanUtils;
-import com.datatorrent.common.codec.JsonStreamCodec;
+import com.google.common.base.Throwables;
import com.datatorrent.api.StreamCodec;
-
-import com.datatorrent.netlet.util.DTThrowable;
+import com.datatorrent.common.codec.JsonStreamCodec;
import com.datatorrent.netlet.util.Slice;
import com.datatorrent.stram.api.StramEvent;
import com.datatorrent.stram.client.EventsAgent;
import com.datatorrent.stram.util.FSPartFileCollection;
import com.datatorrent.stram.util.SharedPubSubWebSocketClient;
+import net.engio.mbassy.listener.Handler;
+
/**
* <p>FSEventRecorder class.</p>
*
@@ -49,7 +52,7 @@ import com.datatorrent.stram.util.SharedPubSubWebSocketClient;
public class FSEventRecorder implements EventRecorder
{
public static final String VERSION = "1.0";
- private final BlockingQueue<StramEvent> queue = new LinkedBlockingQueue<StramEvent>();
+ private final BlockingQueue<StramEvent> queue = new LinkedBlockingQueue<>();
private static final Logger LOG = LoggerFactory.getLogger(FSEventRecorder.class);
private FSPartFileCollection storage;
private String basePath = ".";
@@ -76,11 +79,9 @@ public class FSEventRecorder implements EventRecorder
wsClient.publish(topic, storage.getLatestIndexLine());
}
}
- }
- catch (InterruptedException ex) {
+ } catch (InterruptedException ex) {
return;
- }
- catch (Exception ex) {
+ } catch (Exception ex) {
LOG.error("Caught Exception", ex);
}
}
@@ -107,7 +108,7 @@ public class FSEventRecorder implements EventRecorder
public void setup()
{
try {
- streamCodec = new JsonStreamCodec<Object>();
+ streamCodec = new JsonStreamCodec<>();
storage = new FSPartFileCollection();
storage.setBasePath(basePath);
storage.setup();
@@ -116,25 +117,14 @@ public class FSEventRecorder implements EventRecorder
if (wsClient != null) {
try {
setupWsClient();
- }
- catch (ExecutionException ex) {
- LOG.error("Cannot connect to gateway at {}", pubSubUrl);
- }
- catch (IOException ex) {
- LOG.error("Cannot connect to gateway at {}", pubSubUrl);
- }
- catch (InterruptedException ex) {
- LOG.error("Cannot connect to gateway at {}", pubSubUrl);
- }
- catch (TimeoutException ex) {
+ } catch (ExecutionException | IOException | InterruptedException | TimeoutException ex) {
LOG.error("Cannot connect to gateway at {}", pubSubUrl);
}
}
eventRecorderThread.start();
- }
- catch (Exception ex) {
- DTThrowable.rethrow(ex);
+ } catch (Exception ex) {
+ throw Throwables.propagate(ex);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/FSRecoveryHandler.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/FSRecoveryHandler.java b/engine/src/main/java/com/datatorrent/stram/FSRecoveryHandler.java
index 99ce5eb..f0af8dc 100644
--- a/engine/src/main/java/com/datatorrent/stram/FSRecoveryHandler.java
+++ b/engine/src/main/java/com/datatorrent/stram/FSRecoveryHandler.java
@@ -18,7 +18,15 @@
*/
package com.datatorrent.stram;
-import java.io.*;
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.ObjectStreamClass;
import java.util.EnumSet;
import org.slf4j.Logger;
@@ -26,8 +34,13 @@ import org.slf4j.LoggerFactory;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.*;
+import org.apache.hadoop.fs.CreateFlag;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Options.Rename;
+import org.apache.hadoop.fs.Path;
import com.datatorrent.stram.util.FSUtil;
@@ -38,9 +51,9 @@ import com.datatorrent.stram.util.FSUtil;
*/
public class FSRecoveryHandler implements StreamingContainerManager.RecoveryHandler
{
- private final static Logger LOG = LoggerFactory.getLogger(FSRecoveryHandler.class);
+ private static final Logger LOG = LoggerFactory.getLogger(FSRecoveryHandler.class);
private final Path basedir;
- private final Path logPath ;
+ private final Path logPath;
private final Path logBackupPath;
private final FileSystem fs;
private final Path snapshotPath;
@@ -90,9 +103,9 @@ public class FSRecoveryHandler implements StreamingContainerManager.RecoveryHand
LOG.info("Creating {}", logPath);
final FSDataOutputStream fsOutputStream;
String scheme = null;
- try{
+ try {
scheme = fs.getScheme();
- }catch(UnsupportedOperationException e){
+ } catch (UnsupportedOperationException e) {
LOG.warn("{} doesn't implement getScheme() method", fs.getClass().getName());
}
if ("file".equals(scheme)) {
@@ -103,7 +116,8 @@ public class FSRecoveryHandler implements StreamingContainerManager.RecoveryHand
fsOutputStream = fs.create(logPath);
}
- DataOutputStream osWrapper = new DataOutputStream(fsOutputStream) {
+ DataOutputStream osWrapper = new DataOutputStream(fsOutputStream)
+ {
@Override
public void flush() throws IOException
{
@@ -154,18 +168,9 @@ public class FSRecoveryHandler implements StreamingContainerManager.RecoveryHand
}
LOG.debug("Writing checkpoint to {}", snapshotPath);
- final FSDataOutputStream fsOutputStream = fs.create(snapshotPath);
- try {
- ObjectOutputStream oos = new ObjectOutputStream(fsOutputStream);
- try {
- oos.writeObject(state);
- }
- finally {
- oos.close();
- }
- }
- finally {
- fsOutputStream.close();
+ try (FSDataOutputStream fsOutputStream = fs.create(snapshotPath);
+ ObjectOutputStream oos = new ObjectOutputStream(fsOutputStream)) {
+ oos.writeObject(state);
}
// remove snapshot backup
if (fs.exists(snapshotBackupPath) && !fs.delete(snapshotBackupPath, false)) {
@@ -192,26 +197,14 @@ public class FSRecoveryHandler implements StreamingContainerManager.RecoveryHand
// combine logs (w/o append, create new file)
Path tmpLogPath = new Path(basedir, "log.combined");
- FSDataOutputStream fsOut = fc.create(tmpLogPath, EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE));
- try {
- FSDataInputStream fsIn = fc.open(logBackupPath);
- try {
+ try (FSDataOutputStream fsOut = fc.create(tmpLogPath, EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE))) {
+ try (FSDataInputStream fsIn = fc.open(logBackupPath)) {
IOUtils.copy(fsIn, fsOut);
}
- finally {
- fsIn.close();
- }
- fsIn = fc.open(logPath);
- try {
+ try (FSDataInputStream fsIn = fc.open(logPath)) {
IOUtils.copy(fsIn, fsOut);
}
- finally {
- fsIn.close();
- }
- }
- finally {
- fsOut.close();
}
fc.rename(tmpLogPath, logPath, Rename.OVERWRITE);
@@ -235,34 +228,26 @@ public class FSRecoveryHandler implements StreamingContainerManager.RecoveryHand
// indeterministic class loading behavior
// http://stackoverflow.com/questions/9110677/readresolve-not-working-an-instance-of-guavas-serializedform-appears
final ClassLoader loader = Thread.currentThread().getContextClassLoader();
- ObjectInputStream ois = new ObjectInputStream(is) {
+ try (ObjectInputStream ois = new ObjectInputStream(is)
+ {
@Override
protected Class<?> resolveClass(ObjectStreamClass objectStreamClass)
- throws IOException, ClassNotFoundException {
- return Class.forName(objectStreamClass.getName(), true, loader);
+ throws IOException, ClassNotFoundException
+ {
+ return Class.forName(objectStreamClass.getName(), true, loader);
}
- };
- //ObjectInputStream ois = new ObjectInputStream(is);
- try {
+ }) {
return ois.readObject();
- }
- catch (ClassNotFoundException cnfe) {
+ } catch (ClassNotFoundException cnfe) {
throw new IOException("Failed to read checkpointed state", cnfe);
}
- finally {
- ois.close();
- }
}
public void writeConnectUri(String uri) throws IOException
{
- DataOutputStream out = fs.create(heartbeatPath, true);
- try {
+ try (DataOutputStream out = fs.create(heartbeatPath, true)) {
out.write(uri.getBytes());
}
- finally {
- out.close();
- }
LOG.debug("Connect address: {} written to {} ", uri, heartbeatPath);
}
@@ -273,8 +258,7 @@ public class FSRecoveryHandler implements StreamingContainerManager.RecoveryHand
DataInputStream in = fs.open(heartbeatPath);
try {
bytes = IOUtils.toByteArray(in);
- }
- finally {
+ } finally {
in.close();
}
@@ -289,8 +273,7 @@ public class FSRecoveryHandler implements StreamingContainerManager.RecoveryHand
{
try {
fs.close();
- }
- finally {
+ } finally {
super.finalize();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/FSStatsRecorder.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/FSStatsRecorder.java b/engine/src/main/java/com/datatorrent/stram/FSStatsRecorder.java
index ea02ce6..74cc533 100644
--- a/engine/src/main/java/com/datatorrent/stram/FSStatsRecorder.java
+++ b/engine/src/main/java/com/datatorrent/stram/FSStatsRecorder.java
@@ -21,17 +21,22 @@ package com.datatorrent.stram;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.lang.reflect.Field;
-import java.util.*;
-import java.util.concurrent.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.datatorrent.common.codec.JsonStreamCodec;
-
import com.datatorrent.api.StreamCodec;
import com.datatorrent.api.annotation.RecordField;
-
+import com.datatorrent.common.codec.JsonStreamCodec;
import com.datatorrent.netlet.util.Slice;
import com.datatorrent.stram.util.FSPartFileCollection;
import com.datatorrent.stram.webapp.ContainerInfo;
@@ -48,13 +53,13 @@ public class FSStatsRecorder implements StatsRecorder
private static final Logger LOG = LoggerFactory.getLogger(FSStatsRecorder.class);
private String basePath = ".";
private FSPartFileCollection containersStorage;
- private final Map<String, FSPartFileCollection> logicalOperatorStorageMap = new ConcurrentHashMap<String, FSPartFileCollection>();
- private final Map<String, Integer> knownContainers = new HashMap<String, Integer>();
- private final Set<String> knownOperators = new HashSet<String>();
+ private final Map<String, FSPartFileCollection> logicalOperatorStorageMap = new ConcurrentHashMap<>();
+ private final Map<String, Integer> knownContainers = new HashMap<>();
+ private final Set<String> knownOperators = new HashSet<>();
private transient StreamCodec<Object> streamCodec;
- private final Map<Class<?>, List<Field>> metaFields = new HashMap<Class<?>, List<Field>>();
- private final Map<Class<?>, List<Field>> statsFields = new HashMap<Class<?>, List<Field>>();
- private final BlockingQueue<WriteOperation> queue = new LinkedBlockingQueue<WriteOperation>();
+ private final Map<Class<?>, List<Field>> metaFields = new HashMap<>();
+ private final Map<Class<?>, List<Field>> statsFields = new HashMap<>();
+ private final BlockingQueue<WriteOperation> queue = new LinkedBlockingQueue<>();
private final StatsRecorderThread statsRecorderThread = new StatsRecorderThread();
private class StatsRecorderThread extends Thread
@@ -67,8 +72,7 @@ public class FSStatsRecorder implements StatsRecorder
WriteOperation wo = queue.take();
if (wo.meta) {
wo.storage.writeMetaData(wo.bytes);
- }
- else {
+ } else {
wo.storage.writeDataItem(wo.bytes, true);
}
Thread.yield();
@@ -78,11 +82,9 @@ public class FSStatsRecorder implements StatsRecorder
operatorStorage.flushData();
}
}
- }
- catch (InterruptedException ex) {
+ } catch (InterruptedException ex) {
return;
- }
- catch (Exception ex) {
+ } catch (Exception ex) {
LOG.error("Caught Exception", ex);
}
}
@@ -90,12 +92,15 @@ public class FSStatsRecorder implements StatsRecorder
}
- private static class WriteOperation {
- WriteOperation(FSPartFileCollection storage, byte[] bytes, boolean meta) {
+ private static class WriteOperation
+ {
+ WriteOperation(FSPartFileCollection storage, byte[] bytes, boolean meta)
+ {
this.storage = storage;
this.bytes = bytes;
this.meta = meta;
}
+
FSPartFileCollection storage;
byte[] bytes;
boolean meta;
@@ -109,14 +114,13 @@ public class FSStatsRecorder implements StatsRecorder
public void setup()
{
try {
- streamCodec = new JsonStreamCodec<Object>();
+ streamCodec = new JsonStreamCodec<>();
containersStorage = new FSPartFileCollection();
containersStorage.setBasePath(basePath + "/containers");
containersStorage.setup();
containersStorage.writeMetaData((VERSION + "\n").getBytes());
statsRecorderThread.start();
- }
- catch (Exception ex) {
+ } catch (Exception ex) {
throw new RuntimeException(ex);
}
}
@@ -157,8 +161,7 @@ public class FSStatsRecorder implements StatsRecorder
bos.write(f.buffer, f.offset, f.length);
bos.write("\n".getBytes());
queue.add(new WriteOperation(containersStorage, bos.toByteArray(), true));
- }
- else {
+ } else {
containerIndex = knownContainers.get(entry.getKey());
}
Map<String, Object> fieldMap = extractRecordFields(containerInfo, "stats");
@@ -183,8 +186,7 @@ public class FSStatsRecorder implements StatsRecorder
operatorStorage.setup();
operatorStorage.writeMetaData((VERSION + "\n").getBytes());
logicalOperatorStorageMap.put(operatorInfo.name, operatorStorage);
- }
- else {
+ } else {
operatorStorage = logicalOperatorStorageMap.get(operatorInfo.name);
}
if (!knownOperators.contains(operatorInfo.id)) {
@@ -209,18 +211,17 @@ public class FSStatsRecorder implements StatsRecorder
public Map<String, Object> extractRecordFields(Object o, String type)
{
- Map<String, Object> fieldMap = new HashMap<String, Object>();
+ Map<String, Object> fieldMap = new HashMap<>();
try {
Map<Class<?>, List<Field>> cacheFields = null;
if (type.equals("meta")) {
cacheFields = metaFields;
- }
- else if (type.equals("stats")) {
+ } else if (type.equals("stats")) {
cacheFields = statsFields;
}
List<Field> fieldList;
if (cacheFields == null || !cacheFields.containsKey(o.getClass())) {
- fieldList = new ArrayList<Field>();
+ fieldList = new ArrayList<>();
for (Class<?> c = o.getClass(); c != Object.class; c = c.getSuperclass()) {
Field[] fields = c.getDeclaredFields();
for (Field field : fields) {
@@ -234,16 +235,14 @@ public class FSStatsRecorder implements StatsRecorder
if (cacheFields != null) {
cacheFields.put(o.getClass(), fieldList);
}
- }
- else {
+ } else {
fieldList = cacheFields.get(o.getClass());
}
for (Field field : fieldList) {
fieldMap.put(field.getName(), field.get(o));
}
- }
- catch (IllegalAccessException ex) {
+ } catch (IllegalAccessException ex) {
throw new RuntimeException(ex);
}
return fieldMap;
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/Journal.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/Journal.java b/engine/src/main/java/com/datatorrent/stram/Journal.java
index 5c4392f..ebd3ab5 100644
--- a/engine/src/main/java/com/datatorrent/stram/Journal.java
+++ b/engine/src/main/java/com/datatorrent/stram/Journal.java
@@ -18,21 +18,23 @@
*/
package com.datatorrent.stram;
-import java.io.*;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import com.esotericsoftware.kryo.KryoException;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.google.common.collect.ImmutableMap;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import com.datatorrent.stram.plan.physical.PTContainer;
import com.datatorrent.stram.plan.physical.PTOperator;
@@ -45,19 +47,18 @@ import com.datatorrent.stram.plan.physical.PTOperator;
*/
public final class Journal
{
- private final static Logger LOG = LoggerFactory.getLogger(Journal.class);
+ private static final Logger LOG = LoggerFactory.getLogger(Journal.class);
private enum RecoverableOperation
{
- OPERATOR_STATE (PTOperator.SET_OPERATOR_STATE),
- CONTAINER_STATE (PTContainer.SET_CONTAINER_STATE),
- OPERATOR_PROPERTY (StreamingContainerManager.SET_OPERATOR_PROPERTY),
- PHYSICAL_OPERATOR_PROPERTY (StreamingContainerManager.SET_PHYSICAL_OPERATOR_PROPERTY);
+ OPERATOR_STATE(PTOperator.SET_OPERATOR_STATE),
+ CONTAINER_STATE(PTContainer.SET_CONTAINER_STATE),
+ OPERATOR_PROPERTY(StreamingContainerManager.SET_OPERATOR_PROPERTY),
+ PHYSICAL_OPERATOR_PROPERTY(StreamingContainerManager.SET_PHYSICAL_OPERATOR_PROPERTY);
private static final Map<Class<? extends Recoverable>, Integer> classToId;
- static
- {
+ static {
final ImmutableMap.Builder<Class<? extends Recoverable>, Integer> builder = ImmutableMap.builder();
for (RecoverableOperation recoverableOperation : RecoverableOperation.values()) {
builder.put(recoverableOperation.operation.getClass(), recoverableOperation.ordinal());
@@ -74,7 +75,7 @@ public final class Journal
private static RecoverableOperation get(int id)
{
- return (id < values().length)? values()[id] : null;
+ return (id < values().length) ? values()[id] : null;
}
private static Integer getId(Class<? extends Recoverable> operationClass)
@@ -86,6 +87,7 @@ public final class Journal
public interface Recoverable
{
void read(Object object, Input in) throws KryoException;
+
void write(Output out) throws KryoException;
}
@@ -96,7 +98,7 @@ public final class Journal
public Journal(StreamingContainerManager scm)
{
this.scm = scm;
- output = new AtomicReference<Output>();
+ output = new AtomicReference<>();
replayMode = new AtomicBoolean(false);
}
@@ -104,9 +106,11 @@ public final class Journal
{
final Output output;
if (out != null) {
- output = new Output(4096, -1) {
+ output = new Output(4096, -1)
+ {
@Override
- public void flush() throws KryoException {
+ public void flush() throws KryoException
+ {
super.flush();
// Kryo does not flush internal output stream during flush. We need to flush it explicitly.
try {
@@ -149,8 +153,7 @@ public final class Journal
op.write(out);
out.flush();
break;
- }
- catch (KryoException e) {
+ } catch (KryoException e) {
// check that no other threads sneaked between get() and synchronized block and set output stream to a new
// stream or null leading to the current stream being closed
if (output.get() == out) {
@@ -195,8 +198,7 @@ public final class Journal
LOG.debug("Done replaying WAL");
replayMode.set(false);
}
- }
- else {
+ } else {
throw new IllegalStateException("Request to replay while journal is already replaying other operations");
}
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/LaunchContainerRunnable.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/LaunchContainerRunnable.java b/engine/src/main/java/com/datatorrent/stram/LaunchContainerRunnable.java
index 863808a..8a8db79 100644
--- a/engine/src/main/java/com/datatorrent/stram/LaunchContainerRunnable.java
+++ b/engine/src/main/java/com/datatorrent/stram/LaunchContainerRunnable.java
@@ -21,9 +21,11 @@ package com.datatorrent.stram;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
-import java.util.*;
-
-import com.google.common.collect.Lists;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,7 +43,11 @@ import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.security.token.TokenIdentifier;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
-import org.apache.hadoop.yarn.api.records.*;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
+import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.api.records.LocalResourceType;
+import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.client.api.async.NMClientAsync;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
@@ -49,10 +55,11 @@ import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;
import org.apache.log4j.DTLoggerFactory;
+import com.google.common.collect.Lists;
+
import com.datatorrent.api.Context;
import com.datatorrent.api.DAG;
import com.datatorrent.api.StreamingApplication;
-
import com.datatorrent.stram.client.StramClientUtils;
import com.datatorrent.stram.engine.StreamingContainer;
import com.datatorrent.stram.plan.logical.LogicalPlan;
@@ -71,7 +78,7 @@ public class LaunchContainerRunnable implements Runnable
{
private static final Logger LOG = LoggerFactory.getLogger(LaunchContainerRunnable.class);
private static final String JAVA_REMOTE_DEBUG_OPTS = "-agentlib:jdwp=transport=dt_socket,server=y,suspend=n";
- private final Map<String, String> containerEnv = new HashMap<String, String>();
+ private final Map<String, String> containerEnv = new HashMap<>();
private final LogicalPlan dag;
private final ByteBuffer tokens;
private final Container container;
@@ -121,7 +128,7 @@ public class LaunchContainerRunnable implements Runnable
public static void addFileToLocalResources(final String name, final FileStatus fileStatus, final LocalResourceType type, final Map<String, LocalResource> localResources)
{
final LocalResource localResource = LocalResource.newInstance(ConverterUtils.getYarnUrlFromPath(fileStatus.getPath()),
- type, LocalResourceVisibility.APPLICATION, fileStatus.getLen(), fileStatus.getModificationTime());
+ type, LocalResourceVisibility.APPLICATION, fileStatus.getLen(), fileStatus.getModificationTime());
localResources.put(name, localResource);
}
@@ -155,13 +162,12 @@ public class LaunchContainerRunnable implements Runnable
ctx.setTokens(tokens);
// Set the local resources
- Map<String, LocalResource> localResources = new HashMap<String, LocalResource>();
+ Map<String, LocalResource> localResources = new HashMap<>();
// add resources for child VM
try {
// child VM dependencies
- FileSystem fs = StramClientUtils.newFileSystemInstance(nmClient.getConfig());
- try {
+ try (FileSystem fs = StramClientUtils.newFileSystemInstance(nmClient.getConfig())) {
addFilesToLocalResources(LocalResourceType.FILE, dag.getAttributes().get(LogicalPlan.LIBRARY_JARS), localResources, fs);
String archives = dag.getAttributes().get(LogicalPlan.ARCHIVES);
if (archives != null) {
@@ -169,11 +175,7 @@ public class LaunchContainerRunnable implements Runnable
}
ctx.setLocalResources(localResources);
}
- finally {
- fs.close();
- }
- }
- catch (IOException e) {
+ } catch (IOException e) {
LOG.error("Failed to prepare local resources.", e);
return;
}
@@ -188,7 +190,7 @@ public class LaunchContainerRunnable implements Runnable
}
LOG.info("Launching on node: {} command: {}", container.getNodeId(), command);
- List<String> commands = new ArrayList<String>();
+ List<String> commands = new ArrayList<>();
commands.add(command.toString());
ctx.setCommands(commands);
@@ -204,13 +206,12 @@ public class LaunchContainerRunnable implements Runnable
public List<CharSequence> getChildVMCommand(String jvmID)
{
- List<CharSequence> vargs = new ArrayList<CharSequence>(8);
+ List<CharSequence> vargs = new ArrayList<>(8);
if (!StringUtils.isBlank(System.getenv(Environment.JAVA_HOME.key()))) {
// node manager provides JAVA_HOME
vargs.add(Environment.JAVA_HOME.$() + "/bin/java");
- }
- else {
+ } else {
vargs.add("java");
}
@@ -219,9 +220,8 @@ public class LaunchContainerRunnable implements Runnable
if (dag.isDebug()) {
vargs.add(JAVA_REMOTE_DEBUG_OPTS);
}
- }
- else {
- Map<String, String> params = new HashMap<String, String>();
+ } else {
+ Map<String, String> params = new HashMap<>();
params.put("applicationId", Integer.toString(container.getId().getApplicationAttemptId().getApplicationId().getId()));
params.put("containerId", Integer.toString(container.getId().getId()));
StrSubstitutor sub = new StrSubstitutor(params, "%(", ")");
@@ -233,18 +233,17 @@ public class LaunchContainerRunnable implements Runnable
List<DAG.OperatorMeta> operatorMetaList = Lists.newArrayList();
int bufferServerMemory = 0;
- for(PTOperator operator: sca.getContainer().getOperators()){
+ for (PTOperator operator : sca.getContainer().getOperators()) {
bufferServerMemory += operator.getBufferServerMemory();
operatorMetaList.add(operator.getOperatorMeta());
}
Context.ContainerOptConfigurator containerOptConfigurator = dag.getAttributes().get(LogicalPlan.CONTAINER_OPTS_CONFIGURATOR);
jvmOpts = containerOptConfigurator.getJVMOptions(operatorMetaList);
- jvmOpts = parseJvmOpts(jvmOpts, ((long) bufferServerMemory) * MB_TO_B);
+ jvmOpts = parseJvmOpts(jvmOpts, ((long)bufferServerMemory) * MB_TO_B);
LOG.info("Jvm opts {} for container {}",jvmOpts,container.getId());
vargs.add(jvmOpts);
- Path childTmpDir = new Path(Environment.PWD.$(),
- YarnConfiguration.DEFAULT_CONTAINER_TEMP_DIR);
+ Path childTmpDir = new Path(Environment.PWD.$(), YarnConfiguration.DEFAULT_CONTAINER_TEMP_DIR);
vargs.add(String.format("-D%s=%s", StreamingContainer.PROP_APP_PATH, dag.assertAppPath()));
vargs.add("-Djava.io.tmpdir=" + childTmpDir);
vargs.add(String.format("-D%scid=%s", StreamingApplication.DT_PREFIX, jvmID));
@@ -266,7 +265,7 @@ public class LaunchContainerRunnable implements Runnable
for (CharSequence str : vargs) {
mergedCommand.append(str).append(" ");
}
- List<CharSequence> vargsFinal = new ArrayList<CharSequence>(1);
+ List<CharSequence> vargsFinal = new ArrayList<>(1);
vargsFinal.add(mergedCommand.toString());
return vargsFinal;
@@ -285,8 +284,7 @@ public class LaunchContainerRunnable implements Runnable
long heapSize = Long.valueOf(split.substring(xmx.length()));
heapSize += memory;
builder.append(xmx).append(heapSize).append(" ");
- }
- else {
+ } else {
builder.append(split).append(" ");
}
}
@@ -303,7 +301,7 @@ public class LaunchContainerRunnable implements Runnable
UserGroupInformation ugi = UserGroupInformation.getLoginUser();
StramDelegationTokenIdentifier identifier = new StramDelegationTokenIdentifier(new Text(ugi.getUserName()), new Text(""), new Text(""));
String service = heartbeatAddress.getAddress().getHostAddress() + ":" + heartbeatAddress.getPort();
- Token<StramDelegationTokenIdentifier> stramToken = new Token<StramDelegationTokenIdentifier>(identifier, delegationTokenManager);
+ Token<StramDelegationTokenIdentifier> stramToken = new Token<>(identifier, delegationTokenManager);
stramToken.setService(new Text(service));
return getTokens(ugi, stramToken);
}
@@ -327,8 +325,7 @@ public class LaunchContainerRunnable implements Runnable
byte[] tokenBytes = dataOutput.getData();
ByteBuffer cTokenBuf = ByteBuffer.wrap(tokenBytes);
return cTokenBuf.duplicate();
- }
- catch (IOException e) {
+ } catch (IOException e) {
throw new RuntimeException("Error generating delegation token", e);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/LocalModeImpl.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/LocalModeImpl.java b/engine/src/main/java/com/datatorrent/stram/LocalModeImpl.java
index d711d9c..e51af6b 100644
--- a/engine/src/main/java/com/datatorrent/stram/LocalModeImpl.java
+++ b/engine/src/main/java/com/datatorrent/stram/LocalModeImpl.java
@@ -31,12 +31,13 @@ import com.datatorrent.stram.plan.logical.LogicalPlanConfiguration;
*
* @since 0.3.2
*/
-public class LocalModeImpl extends LocalMode {
-
+public class LocalModeImpl extends LocalMode
+{
private final LogicalPlan lp = new LogicalPlan();
@Override
- public DAG getDAG() {
+ public DAG getDAG()
+ {
return lp;
}
@@ -62,7 +63,8 @@ public class LocalModeImpl extends LocalMode {
}
@Override
- public Controller getController() {
+ public Controller getController()
+ {
try {
return new StramLocalCluster(lp);
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/RecoverableRpcProxy.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/RecoverableRpcProxy.java b/engine/src/main/java/com/datatorrent/stram/RecoverableRpcProxy.java
index 97fce63..1de581e 100644
--- a/engine/src/main/java/com/datatorrent/stram/RecoverableRpcProxy.java
+++ b/engine/src/main/java/com/datatorrent/stram/RecoverableRpcProxy.java
@@ -22,28 +22,29 @@ import java.io.Closeable;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
+import java.net.ConnectException;
import java.net.InetSocketAddress;
+import java.net.SocketTimeoutException;
import java.net.URI;
import java.nio.charset.Charset;
+import java.util.List;
-import static java.lang.Thread.sleep;
-
-import org.apache.http.NameValuePair;
-import org.apache.http.client.utils.URIBuilder;
-import org.apache.http.client.utils.URLEncodedUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.http.NameValuePair;
+import org.apache.http.client.utils.URIBuilder;
+import org.apache.http.client.utils.URLEncodedUtils;
+
+import com.google.common.base.Throwables;
-import com.datatorrent.netlet.util.DTThrowable;
import com.datatorrent.stram.api.StreamingContainerUmbilicalProtocol;
-import java.net.ConnectException;
-import java.net.SocketTimeoutException;
-import java.util.List;
+import static java.lang.Thread.sleep;
/**
* Heartbeat RPC proxy invocation handler that handles fail over.
@@ -105,11 +106,9 @@ public class RecoverableRpcProxy implements java.lang.reflect.InvocationHandler,
String key = pair.getName();
if (QP_rpcTimeout.equals(key)) {
this.rpcTimeout = Integer.parseInt(value);
- }
- else if (QP_retryTimeoutMillis.equals(key)) {
+ } else if (QP_retryTimeoutMillis.equals(key)) {
this.retryTimeoutMillis = Long.parseLong(value);
- }
- else if (QP_retryDelayMillis.equals(key)) {
+ } else if (QP_retryDelayMillis.equals(key)) {
this.retryDelayMillis = Long.parseLong(value);
}
}
@@ -120,7 +119,7 @@ public class RecoverableRpcProxy implements java.lang.reflect.InvocationHandler,
public StreamingContainerUmbilicalProtocol getProxy()
{
- StreamingContainerUmbilicalProtocol recoverableProxy = (StreamingContainerUmbilicalProtocol) java.lang.reflect.Proxy.newProxyInstance(umbilical.getClass().getClassLoader(), umbilical.getClass().getInterfaces(), this);
+ StreamingContainerUmbilicalProtocol recoverableProxy = (StreamingContainerUmbilicalProtocol)java.lang.reflect.Proxy.newProxyInstance(umbilical.getClass().getClassLoader(), umbilical.getClass().getInterfaces(), this);
return recoverableProxy;
}
@@ -129,7 +128,7 @@ public class RecoverableRpcProxy implements java.lang.reflect.InvocationHandler,
public Object invoke(Object proxy, Method method, Object[] args) throws ConnectException, SocketTimeoutException, InterruptedException, IllegalAccessException
{
Object result;
- for (; ; ) {
+ while (true) {
try {
if (umbilical == null) {
connect();
@@ -140,8 +139,7 @@ public class RecoverableRpcProxy implements java.lang.reflect.InvocationHandler,
//long end = System.nanoTime();
//LOG.info(String.format("%s took %d ns", method.getName(), (end - start)));
return result;
- }
- catch (InvocationTargetException e) {
+ } catch (InvocationTargetException e) {
// handle RPC failure
Throwable targetException = e.getTargetException();
long connectMillis = System.currentTimeMillis() - lastCompletedCallTms;
@@ -149,21 +147,17 @@ public class RecoverableRpcProxy implements java.lang.reflect.InvocationHandler,
LOG.warn("RPC failure, attempting reconnect after {} ms (remaining {} ms)", retryDelayMillis, retryTimeoutMillis - connectMillis, targetException);
close();
sleep(retryDelayMillis);
- }
- else {
+ } else {
LOG.error("Giving up RPC connection recovery after {} ms", connectMillis, targetException);
if (targetException instanceof java.net.ConnectException) {
- throw (java.net.ConnectException) targetException;
- }
- else if (targetException instanceof java.net.SocketTimeoutException) {
- throw (java.net.SocketTimeoutException) targetException;
- }
- else {
- DTThrowable.rethrow(targetException);
+ throw (java.net.ConnectException)targetException;
+ } else if (targetException instanceof java.net.SocketTimeoutException) {
+ throw (java.net.SocketTimeoutException)targetException;
+ } else {
+ throw Throwables.propagate(targetException);
}
}
- }
- catch (IOException ex) {
+ } catch (IOException ex) {
close();
throw new RuntimeException(ex);
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/ResourceRequestHandler.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/ResourceRequestHandler.java b/engine/src/main/java/com/datatorrent/stram/ResourceRequestHandler.java
index 4bffd62..6ecc7c5 100644
--- a/engine/src/main/java/com/datatorrent/stram/ResourceRequestHandler.java
+++ b/engine/src/main/java/com/datatorrent/stram/ResourceRequestHandler.java
@@ -23,6 +23,9 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.tuple.MutablePair;
import org.apache.hadoop.yarn.api.records.NodeReport;
@@ -32,8 +35,6 @@ import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.client.api.AMRMClient;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.util.Records;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
@@ -52,8 +53,8 @@ import com.datatorrent.stram.plan.physical.PTOperator.HostOperatorSet;
public class ResourceRequestHandler
{
- private final static Logger LOG = LoggerFactory.getLogger(ResourceRequestHandler.class);
- private final static String INVALID_HOST = "INVALID_HOST";
+ private static final Logger LOG = LoggerFactory.getLogger(ResourceRequestHandler.class);
+ private static final String INVALID_HOST = "INVALID_HOST";
protected static final int NUMBER_MISSED_HEARTBEATS = 30;
@@ -274,8 +275,8 @@ public class ResourceRequestHandler
/**
* Populate list of nodes where container cannot be allocated due to anti-affinity constraints
- * @param container
- * @param List of nodes where container cannot be allocated
+ * @param c container
+ * @param antiHosts List of nodes where container cannot be allocated
*/
public void populateAntiHostList(PTContainer c, List<String> antiHosts)
{
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/StramAppContext.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/StramAppContext.java b/engine/src/main/java/com/datatorrent/stram/StramAppContext.java
index 6b36e82..2997992 100644
--- a/engine/src/main/java/com/datatorrent/stram/StramAppContext.java
+++ b/engine/src/main/java/com/datatorrent/stram/StramAppContext.java
@@ -18,18 +18,18 @@
*/
package com.datatorrent.stram;
-import com.datatorrent.stram.api.AppDataSource;
+import java.util.List;
+import java.util.Map;
+
import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.yarn.util.Clock;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.util.Clock;
import com.datatorrent.api.Attribute.AttributeMap.AttributeInitializer;
import com.datatorrent.api.Context;
-
+import com.datatorrent.stram.api.AppDataSource;
import com.datatorrent.stram.webapp.AppInfo;
-import java.util.List;
-import java.util.Map;
/**
*
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/StramClient.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/StramClient.java b/engine/src/main/java/com/datatorrent/stram/StramClient.java
index 56cc6dc..c1dfffd 100644
--- a/engine/src/main/java/com/datatorrent/stram/StramClient.java
+++ b/engine/src/main/java/com/datatorrent/stram/StramClient.java
@@ -153,7 +153,7 @@ public class StramClient
};
private static final Class<?>[] DATATORRENT_SECURITY_CLASSES =
- (Class<?>[]) ArrayUtils.addAll(DATATORRENT_CLASSES, DATATORRENT_SECURITY_SPECIFIC_CLASSES);
+ (Class<?>[])ArrayUtils.addAll(DATATORRENT_CLASSES, DATATORRENT_SECURITY_SPECIFIC_CLASSES);
public StramClient(Configuration conf, LogicalPlan dag) throws Exception
{
@@ -175,14 +175,13 @@ public class StramClient
public static LinkedHashSet<String> findJars(LogicalPlan dag, Class<?>[] defaultClasses)
{
- List<Class<?>> jarClasses = new ArrayList<Class<?>>();
+ List<Class<?>> jarClasses = new ArrayList<>();
for (String className : dag.getClassNames()) {
try {
Class<?> clazz = Thread.currentThread().getContextClassLoader().loadClass(className);
jarClasses.add(clazz);
- }
- catch (ClassNotFoundException e) {
+ } catch (ClassNotFoundException e) {
throw new IllegalArgumentException("Failed to load class " + className, e);
}
}
@@ -202,8 +201,8 @@ public class StramClient
LOG.debug("Deploy dependencies: {}", jarClasses);
}
- LinkedHashSet<String> localJarFiles = new LinkedHashSet<String>(); // avoid duplicates
- HashMap<String, String> sourceToJar = new HashMap<String, String>();
+ LinkedHashSet<String> localJarFiles = new LinkedHashSet<>(); // avoid duplicates
+ HashMap<String, String> sourceToJar = new HashMap<>();
for (Class<?> jarClass : jarClasses) {
if (jarClass.getProtectionDomain().getCodeSource() == null) {
@@ -245,15 +244,13 @@ public class StramClient
URI localFileURI = null;
try {
localFileURI = new URI(localFile);
- }
- catch (URISyntaxException e) {
+ } catch (URISyntaxException e) {
throw new IOException(e);
}
if (localFileURI.getScheme() == null || localFileURI.getScheme().startsWith("file")) {
LOG.info("Copy {} from local filesystem to {}", localFile, dst);
fs.copyFromLocalFile(false, true, src, dst);
- }
- else {
+ } else {
LOG.info("Copy {} from DFS to {}", localFile, dst);
FileUtil.copy(fs, src, fs, dst, false, true, conf);
}
@@ -279,7 +276,7 @@ public class StramClient
InputStream logIs = recoveryHandler.getLog();
// modify snapshot state to switch app id
- ((StreamingContainerManager.CheckpointState) snapshot).setApplicationId(this.dag, conf);
+ ((StreamingContainerManager.CheckpointState)snapshot).setApplicationId(this.dag, conf);
Path checkpointPath = new Path(newAppDir, LogicalPlan.SUBDIR_CHECKPOINTS);
FileSystem fs = FileSystem.newInstance(origAppDir.toUri(), conf);
@@ -304,8 +301,7 @@ public class StramClient
LOG.debug("Copying {} to {}", f.getPath(), targetPath);
FileUtil.copy(fs, f.getPath(), fs, new Path(targetPath), false, conf);
//FSUtil.copy(fs, f, fs, new Path(targetPath), false, false, conf);
- }
- else {
+ } else {
LOG.debug("Ignoring {} as it already exists under {}", f.getPath(), targetPath);
//FSUtil.setPermission(fs, new Path(targetPath), new FsPermission((short)0777));
}
@@ -324,16 +320,14 @@ public class StramClient
{
Class<?>[] defaultClasses;
- if(applicationType.equals(YARN_APPLICATION_TYPE)) {
+ if (applicationType.equals(YARN_APPLICATION_TYPE)) {
//TODO restrict the security check to only check if security is enabled for webservices.
- if(UserGroupInformation.isSecurityEnabled()) {
+ if (UserGroupInformation.isSecurityEnabled()) {
defaultClasses = DATATORRENT_SECURITY_CLASSES;
- }
- else {
+ } else {
defaultClasses = DATATORRENT_CLASSES;
}
- }
- else {
+ } else {
throw new IllegalStateException(applicationType + " is not a valid application type.");
}
@@ -344,8 +338,7 @@ public class StramClient
}
YarnClusterMetrics clusterMetrics = yarnClient.getYarnClusterMetrics();
- LOG.info("Got Cluster metric info from ASM"
- + ", numNodeManagers=" + clusterMetrics.getNumNodeManagers());
+ LOG.info("Got Cluster metric info from ASM, numNodeManagers={}", clusterMetrics.getNumNodeManagers());
//GetClusterNodesRequest clusterNodesReq = Records.newRecord(GetClusterNodesRequest.class);
//GetClusterNodesResponse clusterNodesResp = rmClient.clientRM.getClusterNodes(clusterNodesReq);
@@ -361,9 +354,7 @@ public class StramClient
List<QueueUserACLInfo> listAclInfo = yarnClient.getQueueAclsInfo();
for (QueueUserACLInfo aclInfo : listAclInfo) {
for (QueueACL userAcl : aclInfo.getUserAcls()) {
- LOG.info("User ACL Info for Queue"
- + ", queueName=" + aclInfo.getQueueName()
- + ", userAcl=" + userAcl.name());
+ LOG.info("User ACL Info for Queue queueName={}, userAcl={}", aclInfo.getQueueName(), userAcl.name());
}
}
@@ -376,9 +367,8 @@ public class StramClient
LOG.info("Max mem capability of resources in this cluster " + maxMem);
int amMemory = dag.getMasterMemoryMB();
if (amMemory > maxMem) {
- LOG.info("AM memory specified above max threshold of cluster. Using max value."
- + ", specified=" + amMemory
- + ", max=" + maxMem);
+ LOG.info("AM memory specified above max threshold of cluster. Using max value, specified={}, max={}",
+ amMemory, maxMem);
amMemory = maxMem;
}
@@ -417,18 +407,14 @@ public class StramClient
}
// For now, only getting tokens for the default file-system.
- FileSystem fs = StramClientUtils.newFileSystemInstance(conf);
- try {
- final Token<?> tokens[] = fs.addDelegationTokens(tokenRenewer, credentials);
+ try (FileSystem fs = StramClientUtils.newFileSystemInstance(conf)) {
+ final Token<?>[] tokens = fs.addDelegationTokens(tokenRenewer, credentials);
if (tokens != null) {
for (Token<?> token : tokens) {
LOG.info("Got dt for " + fs.getUri() + "; " + token);
}
}
}
- finally {
- fs.close();
- }
new ClientRMHelper(yarnClient, conf).addRMDelegationToken(tokenRenewer, credentials);
@@ -441,11 +427,10 @@ public class StramClient
// set local resources for the application master
// local files or archives as needed
// In this scenario, the jar file for the application master is part of the local resources
- Map<String, LocalResource> localResources = new HashMap<String, LocalResource>();
+ Map<String, LocalResource> localResources = new HashMap<>();
// copy required jar files to dfs, to be localized for containers
- FileSystem fs = StramClientUtils.newFileSystemInstance(conf);
- try {
+ try (FileSystem fs = StramClientUtils.newFileSystemInstance(conf)) {
Path appsBasePath = new Path(StramClientUtils.getDTDFSRootDir(fs, conf), StramClientUtils.SUBDIR_APPS);
Path appPath;
String configuredAppPath = dag.getValue(LogicalPlan.APPLICATION_PATH);
@@ -488,8 +473,8 @@ public class StramClient
dag.setAttribute(OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(checkpointPath.toString(), conf));
}
- if(dag.getAttributes().get(LogicalPlan.CONTAINER_OPTS_CONFIGURATOR) == null){
- dag.setAttribute(LogicalPlan.CONTAINER_OPTS_CONFIGURATOR,new BasicContainerOptConfigurator());
+ if (dag.getAttributes().get(LogicalPlan.CONTAINER_OPTS_CONFIGURATOR) == null) {
+ dag.setAttribute(LogicalPlan.CONTAINER_OPTS_CONFIGURATOR, new BasicContainerOptConfigurator());
}
// Set the log4j properties if needed
@@ -532,7 +517,7 @@ public class StramClient
//amContainer.setContainerTokens(containerToken);
// Set the env variables to be setup in the env where the application master will be run
LOG.info("Set the environment for the application master");
- Map<String, String> env = new HashMap<String, String>();
+ Map<String, String> env = new HashMap<>();
// Add application jar(s) location to classpath
// At some point we should not be required to add
@@ -559,7 +544,7 @@ public class StramClient
amContainer.setEnvironment(env);
// Set the necessary command to execute the application master
- ArrayList<CharSequence> vargs = new ArrayList<CharSequence>(30);
+ ArrayList<CharSequence> vargs = new ArrayList<>(30);
// Set java executable command
LOG.info("Setting up app master command");
@@ -572,8 +557,7 @@ public class StramClient
if (dag.getMasterJVMOptions() != null) {
vargs.add(dag.getMasterJVMOptions());
}
- Path tmpDir = new Path(ApplicationConstants.Environment.PWD.$(),
- YarnConfiguration.DEFAULT_CONTAINER_TEMP_DIR);
+ Path tmpDir = new Path(ApplicationConstants.Environment.PWD.$(), YarnConfiguration.DEFAULT_CONTAINER_TEMP_DIR);
vargs.add("-Djava.io.tmpdir=" + tmpDir);
vargs.add("-Xmx" + (amMemory * 3 / 4) + "m");
vargs.add("-XX:+HeapDumpOnOutOfMemoryError");
@@ -600,7 +584,7 @@ public class StramClient
}
LOG.info("Completed setting up app master command " + command.toString());
- List<String> commands = new ArrayList<String>();
+ List<String> commands = new ArrayList<>();
commands.add(command.toString());
amContainer.setCommands(commands);
@@ -627,20 +611,17 @@ public class StramClient
// Ignore the response as either a valid response object is returned on success
// or an exception thrown to denote some form of a failure
String specStr = Objects.toStringHelper("Submitting application: ")
- .add("name", appContext.getApplicationName())
- .add("queue", appContext.getQueue())
- .add("user", UserGroupInformation.getLoginUser())
- .add("resource", appContext.getResource())
- .toString();
+ .add("name", appContext.getApplicationName())
+ .add("queue", appContext.getQueue())
+ .add("user", UserGroupInformation.getLoginUser())
+ .add("resource", appContext.getResource())
+ .toString();
LOG.info(specStr);
if (dag.isDebug()) {
//LOG.info("Full submission context: " + appContext);
}
yarnClient.submitApplication(appContext);
}
- finally {
- fs.close();
- }
}
public ApplicationReport getApplicationReport() throws YarnException, IOException
@@ -672,18 +653,12 @@ public class StramClient
@Override
public boolean exitLoop(ApplicationReport report)
{
- LOG.info("Got application report from ASM for"
- + ", appId=" + appId.getId()
- + ", clientToken=" + report.getClientToAMToken()
- + ", appDiagnostics=" + report.getDiagnostics()
- + ", appMasterHost=" + report.getHost()
- + ", appQueue=" + report.getQueue()
- + ", appMasterRpcPort=" + report.getRpcPort()
- + ", appStartTime=" + report.getStartTime()
- + ", yarnAppState=" + report.getYarnApplicationState().toString()
- + ", distributedFinalState=" + report.getFinalApplicationStatus().toString()
- + ", appTrackingUrl=" + report.getTrackingUrl()
- + ", appUser=" + report.getUser());
+ LOG.info("Got application report from ASM for, appId={}, clientToken={}, appDiagnostics={}, appMasterHost={}," +
+ "appQueue={}, appMasterRpcPort={}, appStartTime={}, yarnAppState={}, distributedFinalState={}, " +
+ "appTrackingUrl={}, appUser={}",
+ appId.getId(), report.getClientToAMToken(), report.getDiagnostics(), report.getHost(),
+ report.getQueue(), report.getRpcPort(), report.getStartTime(), report.getYarnApplicationState(),
+ report.getFinalApplicationStatus(), report.getTrackingUrl(), report.getUser());
return false;
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/StramLocalCluster.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/StramLocalCluster.java b/engine/src/main/java/com/datatorrent/stram/StramLocalCluster.java
index d0a6ed9..b3cf63f 100644
--- a/engine/src/main/java/com/datatorrent/stram/StramLocalCluster.java
+++ b/engine/src/main/java/com/datatorrent/stram/StramLocalCluster.java
@@ -39,14 +39,12 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.ipc.ProtocolSignature;
import org.apache.hadoop.net.NetUtils;
-import com.datatorrent.api.Context;
import com.datatorrent.api.DAG;
import com.datatorrent.api.LocalMode.Controller;
import com.datatorrent.api.Operator;
import com.datatorrent.bufferserver.server.Server;
import com.datatorrent.bufferserver.storage.DiskStorage;
import com.datatorrent.common.util.AsyncFSStorageAgent;
-import com.datatorrent.common.util.FSStorageAgent;
import com.datatorrent.stram.StreamingContainerAgent.ContainerStartRequest;
import com.datatorrent.stram.StreamingContainerManager.ContainerResource;
import com.datatorrent.stram.api.StreamingContainerUmbilicalProtocol;
@@ -76,10 +74,10 @@ public class StramLocalCluster implements Runnable, Controller
private InetSocketAddress bufferServerAddress;
private boolean perContainerBufferServer;
private Server bufferServer = null;
- private final Map<String, LocalStreamingContainer> childContainers = new ConcurrentHashMap<String, LocalStreamingContainer>();
+ private final Map<String, LocalStreamingContainer> childContainers = new ConcurrentHashMap<>();
private int containerSeq = 0;
private boolean appDone = false;
- private final Map<String, StreamingContainer> injectShutdown = new ConcurrentHashMap<String, StreamingContainer>();
+ private final Map<String, StreamingContainer> injectShutdown = new ConcurrentHashMap<>();
private boolean heartbeatMonitoringEnabled = true;
private Callable<Boolean> exitCondition;
@@ -100,8 +98,7 @@ public class StramLocalCluster implements Runnable, Controller
}
@Override
- public ProtocolSignature getProtocolSignature(String protocol,
- long clientVersion, int clientMethodsHash) throws IOException
+ public ProtocolSignature getProtocolSignature(String protocol, long clientVersion, int clientMethodsHash) throws IOException
{
throw new UnsupportedOperationException("not implemented in local mode");
}
@@ -111,8 +108,7 @@ public class StramLocalCluster implements Runnable, Controller
{
try {
log(containerId, msg);
- }
- catch (IOException ex) {
+ } catch (IOException ex) {
// ignore
}
}
@@ -148,8 +144,7 @@ public class StramLocalCluster implements Runnable, Controller
rsp = SerializationUtils.clone(rsp);
}
return rsp;
- }
- finally {
+ } finally {
LocalStreamingContainer c = childContainers.get(msg.getContainerId());
synchronized (c.heartbeatCount) {
c.heartbeatCount.incrementAndGet();
@@ -183,13 +178,11 @@ public class StramLocalCluster implements Runnable, Controller
// main thread enters heartbeat loop
stramChild.heartbeatLoop();
hasError = false;
- }
- finally {
+ } finally {
// teardown
try {
stramChild.teardown();
- }
- catch (Exception e) {
+ } catch (Exception e) {
if (!hasError) {
throw e;
}
@@ -269,12 +262,10 @@ public class StramLocalCluster implements Runnable, Controller
try {
StreamingContainerContext ctx = umbilical.getInitContext(containerId);
LocalStreamingContainer.run(child, ctx);
- }
- catch (Exception e) {
+ } catch (Exception e) {
LOG.error("Container {} failed", containerId, e);
throw new RuntimeException(e);
- }
- finally {
+ } finally {
childContainers.remove(containerId);
LOG.info("Container {} terminating.", containerId);
}
@@ -292,11 +283,9 @@ public class StramLocalCluster implements Runnable, Controller
String pathUri = CLUSTER_WORK_DIR.toURI().toString();
try {
FileContext.getLocalFSFileContext().delete(new Path(pathUri/*CLUSTER_WORK_DIR.getAbsolutePath()*/), true);
- }
- catch (IllegalArgumentException e) {
+ } catch (IllegalArgumentException e) {
throw e;
- }
- catch (IOException e) {
+ } catch (IOException e) {
throw new RuntimeException("could not cleanup test dir", e);
}
@@ -498,8 +487,7 @@ public class StramLocalCluster implements Runnable, Controller
if (!appDone) {
try {
Thread.sleep(1000);
- }
- catch (InterruptedException e) {
+ } catch (InterruptedException e) {
LOG.info("Sleep interrupted " + e.getMessage());
break;
}