You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nemo.apache.org by wo...@apache.org on 2019/10/15 13:54:16 UTC
[incubator-nemo] branch master updated: [NEMO-424] Fix Sonarcloud
bugs regarding Optional (#241)
This is an automated email from the ASF dual-hosted git repository.
wonook pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
The following commit(s) were added to refs/heads/master by this push:
new 629b1a4 [NEMO-424] Fix Sonarcloud bugs regarding Optional (#241)
629b1a4 is described below
commit 629b1a4c514c5edb3c7e82111586765ab66042fd
Author: Sanha Lee <sa...@gmail.com>
AuthorDate: Tue Oct 15 22:54:08 2019 +0900
[NEMO-424] Fix Sonarcloud bugs regarding Optional (#241)
JIRA: [NEMO-424: Fix Sonarcloud bugs regarding Optional](https://issues.apache.org/jira/projects/NEMO/issues/NEMO-424)
**Major changes:**
- Fixes Sonarcloud bugs regarding Optional
**Minor changes to note:**
- Fixes other Sonarcloud bugs assigned to johnyangk
- Fixes other Sonarcloud bugs assigned to sanha
closes #241
---
.../src/main/java/org/apache/nemo/common/Util.java | 5 ++-
.../nemo/common/partitioner/Partitioner.java | 12 ++++--
.../compiletime/annotating/ResourceSitePass.java | 2 +-
.../reshaping/SamplingSkewReshapingPass.java | 4 +-
.../compiletime/reshaping/SkewHandlingUtil.java | 6 ++-
.../optimizer/pass/runtime/SkewRunTimePass.java | 5 ++-
.../common/message/ncs/NcsMessageEnvironment.java | 32 ++++++++++------
.../nemo/runtime/common/plan/StagePartitioner.java | 2 +-
.../java/org/apache/nemo/driver/NemoContext.java | 3 +-
.../bytetransfer/ClosableBlockingQueue.java | 2 +-
.../nemo/runtime/executor/data/PipeContainer.java | 1 +
.../data/partition/SerializedPartition.java | 2 +-
.../executor/datatransfer/BlockInputReader.java | 44 ++++++++++++----------
.../executor/datatransfer/BlockOutputWriter.java | 2 +-
.../executor/datatransfer/PipeInputReader.java | 9 +++--
.../executor/datatransfer/PipeOutputWriter.java | 19 ++++++----
.../task/MultiThreadParentTaskDataFetcher.java | 2 +-
.../nemo/runtime/executor/task/TaskExecutor.java | 1 +
.../apache/nemo/runtime/master/BlockMetadata.java | 2 +-
.../nemo/runtime/master/PipeManagerMaster.java | 1 +
.../runtime/master/resource/ContainerManager.java | 5 ++-
21 files changed, 97 insertions(+), 64 deletions(-)
diff --git a/common/src/main/java/org/apache/nemo/common/Util.java b/common/src/main/java/org/apache/nemo/common/Util.java
index 9216b01..72e9200 100644
--- a/common/src/main/java/org/apache/nemo/common/Util.java
+++ b/common/src/main/java/org/apache/nemo/common/Util.java
@@ -117,7 +117,10 @@ public final class Util {
final IRVertex newSrc,
final IRVertex newDst) {
return cloneEdge(
- edgeToClone.getPropertyValue(CommunicationPatternProperty.class).get(), edgeToClone, newSrc, newDst);
+ edgeToClone.getPropertyValue(CommunicationPatternProperty.class).orElseThrow(IllegalStateException::new),
+ edgeToClone,
+ newSrc,
+ newDst);
}
/**
diff --git a/common/src/main/java/org/apache/nemo/common/partitioner/Partitioner.java b/common/src/main/java/org/apache/nemo/common/partitioner/Partitioner.java
index fcc077c..f10d935 100644
--- a/common/src/main/java/org/apache/nemo/common/partitioner/Partitioner.java
+++ b/common/src/main/java/org/apache/nemo/common/partitioner/Partitioner.java
@@ -54,7 +54,7 @@ public interface Partitioner<K extends Serializable> {
static Partitioner getPartitioner(final ExecutionPropertyMap<EdgeExecutionProperty> edgeProperties,
final ExecutionPropertyMap<VertexExecutionProperty> dstProperties) {
final PartitionerProperty.Type type =
- edgeProperties.get(PartitionerProperty.class).get().left();
+ edgeProperties.get(PartitionerProperty.class).orElseThrow(IllegalStateException::new).left();
final Partitioner partitioner;
switch (type) {
case INTACT:
@@ -64,11 +64,15 @@ public interface Partitioner<K extends Serializable> {
partitioner = new DedicatedKeyPerElementPartitioner();
break;
case HASH:
- final int numOfPartitions = edgeProperties.get(PartitionerProperty.class).get().right();
+ final int numOfPartitions = edgeProperties
+ .get(PartitionerProperty.class)
+ .orElseThrow(IllegalStateException::new)
+ .right();
final int actualNumOfPartitions = (numOfPartitions == PartitionerProperty.NUM_EQUAL_TO_DST_PARALLELISM)
- ? dstProperties.get(ParallelismProperty.class).get()
+ ? dstProperties.get(ParallelismProperty.class).orElseThrow(IllegalStateException::new)
: numOfPartitions;
- final KeyExtractor keyExtractor = edgeProperties.get(KeyExtractorProperty.class).get();
+ final KeyExtractor keyExtractor = edgeProperties.get(KeyExtractorProperty.class)
+ .orElseThrow(IllegalStateException::new);
partitioner = new HashPartitioner(actualNumOfPartitions, keyExtractor);
break;
default:
diff --git a/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/annotating/ResourceSitePass.java b/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/annotating/ResourceSitePass.java
index 207c894..bb97bc7 100644
--- a/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/annotating/ResourceSitePass.java
+++ b/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/annotating/ResourceSitePass.java
@@ -164,7 +164,7 @@ public final class ResourceSitePass extends AnnotatingPass {
*/
private static boolean isOneToOneEdge(final Collection<IREdge> inEdges) {
return inEdges.size() == 1 && inEdges.iterator().next()
- .getPropertyValue(CommunicationPatternProperty.class).get()
+ .getPropertyValue(CommunicationPatternProperty.class).orElseThrow(IllegalStateException::new)
.equals(CommunicationPatternProperty.Value.ONE_TO_ONE);
}
diff --git a/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SamplingSkewReshapingPass.java b/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SamplingSkewReshapingPass.java
index 1da832b..a90c1f8 100644
--- a/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SamplingSkewReshapingPass.java
+++ b/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SamplingSkewReshapingPass.java
@@ -130,9 +130,9 @@ public final class SamplingSkewReshapingPass extends ReshapingPass {
unionSet.add(curVertex);
for (final IREdge inEdge : dag.getIncomingEdgesOf(curVertex)) {
if (CommunicationPatternProperty.Value.ONE_TO_ONE
- .equals(inEdge.getPropertyValue(CommunicationPatternProperty.class).get())
+ .equals(inEdge.getPropertyValue(CommunicationPatternProperty.class).orElseThrow(IllegalStateException::new))
&& DataStoreProperty.Value.MEMORY_STORE
- .equals(inEdge.getPropertyValue(DataStoreProperty.class).get())
+ .equals(inEdge.getPropertyValue(DataStoreProperty.class).orElseThrow(IllegalStateException::new))
&& dag.getIncomingEdgesOf(curVertex).size() == 1) {
unionSet.addAll(recursivelyBuildPartition(inEdge.getSrc(), dag));
}
diff --git a/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SkewHandlingUtil.java b/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SkewHandlingUtil.java
index b724bdd..8e9f597 100644
--- a/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SkewHandlingUtil.java
+++ b/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SkewHandlingUtil.java
@@ -68,11 +68,13 @@ final class SkewHandlingUtil {
static EncoderProperty getEncoder(final IREdge irEdge) {
return EncoderProperty.of(PairEncoderFactory
- .of(irEdge.getPropertyValue(KeyEncoderProperty.class).get(), LongEncoderFactory.of()));
+ .of(irEdge.getPropertyValue(KeyEncoderProperty.class).orElseThrow(IllegalStateException::new),
+ LongEncoderFactory.of()));
}
static DecoderProperty getDecoder(final IREdge irEdge) {
return DecoderProperty.of(PairDecoderFactory
- .of(irEdge.getPropertyValue(KeyDecoderProperty.class).get(), LongDecoderFactory.of()));
+ .of(irEdge.getPropertyValue(KeyDecoderProperty.class).orElseThrow(IllegalStateException::new),
+ LongDecoderFactory.of()));
}
}
diff --git a/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/runtime/SkewRunTimePass.java b/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/runtime/SkewRunTimePass.java
index 54e4599..7502b16 100644
--- a/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/runtime/SkewRunTimePass.java
+++ b/compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/runtime/SkewRunTimePass.java
@@ -65,8 +65,9 @@ public final class SkewRunTimePass extends RunTimePass<Map<Object, Long>> {
// Use the following execution properties.
final Pair<PartitionerProperty.Type, Integer> partitionerProperty =
- representativeEdge.getPropertyValue(PartitionerProperty.class).get();
- final int dstParallelism = representativeEdge.getDst().getPropertyValue(ParallelismProperty.class).get();
+ representativeEdge.getPropertyValue(PartitionerProperty.class).orElseThrow(IllegalStateException::new);
+ final int dstParallelism = representativeEdge.getDst().getPropertyValue(ParallelismProperty.class)
+ .orElseThrow(IllegalStateException::new);
// Compute the optimal partition distribution, using the message value.
final Map<Object, Long> messageValue = message.getMessageValue();
diff --git a/runtime/common/src/main/java/org/apache/nemo/runtime/common/message/ncs/NcsMessageEnvironment.java b/runtime/common/src/main/java/org/apache/nemo/runtime/common/message/ncs/NcsMessageEnvironment.java
index 5a45e0c..4986aa2 100644
--- a/runtime/common/src/main/java/org/apache/nemo/runtime/common/message/ncs/NcsMessageEnvironment.java
+++ b/runtime/common/src/main/java/org/apache/nemo/runtime/common/message/ncs/NcsMessageEnvironment.java
@@ -91,21 +91,29 @@ public final class NcsMessageEnvironment implements MessageEnvironment {
@Override
public <T> Future<MessageSender<T>> asyncConnect(final String receiverId, final String listenerId) {
- try {
- // If the connection toward the receiver exists already, reuses it.
- final Connection connection;
- if (receiverToConnectionMap.containsKey(receiverId)) {
- connection = receiverToConnectionMap.get(receiverId);
- } else {
- connection = connectionFactory.newConnection(idFactory.getNewInstance(receiverId));
+ // If the connection toward the receiver exists already, reuses it.
+ final Connection connection;
+
+ if (receiverToConnectionMap.containsKey(receiverId)) {
+ connection = receiverToConnectionMap.get(receiverId);
+ } else {
+ connection = connectionFactory.newConnection(idFactory.getNewInstance(receiverId));
+ try {
connection.open();
+ } catch (final NetworkException e) {
+ try {
+ connection.close();
+ } catch (final NetworkException exceptionToIgnore) {
+ LOG.info("Can't close the broken connection.", exceptionToIgnore);
+ }
+
+ final CompletableFuture<MessageSender<T>> failedFuture = new CompletableFuture<>();
+ failedFuture.completeExceptionally(e);
+ return failedFuture;
}
- return CompletableFuture.completedFuture((MessageSender) new NcsMessageSender(connection, replyFutureMap));
- } catch (final NetworkException e) {
- final CompletableFuture<MessageSender<T>> failedFuture = new CompletableFuture<>();
- failedFuture.completeExceptionally(e);
- return failedFuture;
}
+
+ return CompletableFuture.completedFuture((MessageSender) new NcsMessageSender(connection, replyFutureMap));
}
@Override
diff --git a/runtime/common/src/main/java/org/apache/nemo/runtime/common/plan/StagePartitioner.java b/runtime/common/src/main/java/org/apache/nemo/runtime/common/plan/StagePartitioner.java
index 43636cd..bc20448 100644
--- a/runtime/common/src/main/java/org/apache/nemo/runtime/common/plan/StagePartitioner.java
+++ b/runtime/common/src/main/java/org/apache/nemo/runtime/common/plan/StagePartitioner.java
@@ -106,7 +106,7 @@ public final class StagePartitioner implements Function<IRDAG, Map<IRVertex, Int
return false;
}
// If the edge is not OneToOne, return false
- if (edge.getPropertyValue(CommunicationPatternProperty.class).get()
+ if (edge.getPropertyValue(CommunicationPatternProperty.class).orElseThrow(IllegalStateException::new)
!= CommunicationPatternProperty.Value.ONE_TO_ONE) {
return false;
}
diff --git a/runtime/driver/src/main/java/org/apache/nemo/driver/NemoContext.java b/runtime/driver/src/main/java/org/apache/nemo/driver/NemoContext.java
index d41b7ce..88c44ca 100644
--- a/runtime/driver/src/main/java/org/apache/nemo/driver/NemoContext.java
+++ b/runtime/driver/src/main/java/org/apache/nemo/driver/NemoContext.java
@@ -44,6 +44,7 @@ public final class NemoContext {
private final Clock clock;
private final int crashTimeSec;
+ private final Random random;
@Inject
private NemoContext(final Executor executor,
@@ -54,6 +55,7 @@ public final class NemoContext {
// For poison handling
this.clock = clock;
this.crashTimeSec = crashTimeSec;
+ this.random = new Random();
}
/**
@@ -87,7 +89,6 @@ public final class NemoContext {
}
private int addNoise(final int number) {
- final Random random = new Random();
final int fiftyPercent = random.nextInt((int) (number * (50.0 / 100.0)));
return random.nextBoolean() ? number + fiftyPercent : number - fiftyPercent; // -50% ~ +50%
}
diff --git a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/bytetransfer/ClosableBlockingQueue.java b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/bytetransfer/ClosableBlockingQueue.java
index 7bc8855..58b7409 100644
--- a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/bytetransfer/ClosableBlockingQueue.java
+++ b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/bytetransfer/ClosableBlockingQueue.java
@@ -33,7 +33,7 @@ public final class ClosableBlockingQueue<T> implements AutoCloseable {
private final Queue<T> queue;
private volatile boolean closed = false;
- private volatile Throwable throwable = null;
+ private volatile Throwable throwable = null; // intentionally volatilize the reference
/**
* Creates a closable blocking queue.
diff --git a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/data/PipeContainer.java b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/data/PipeContainer.java
index 31c868e..b50788b 100644
--- a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/data/PipeContainer.java
+++ b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/data/PipeContainer.java
@@ -77,6 +77,7 @@ public final class PipeContainer {
}
return new ArrayList<>(indexToValue.values());
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw new RuntimeException(e);
} finally {
lock.unlock();
diff --git a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/data/partition/SerializedPartition.java b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/data/partition/SerializedPartition.java
index a2abbdb..2606c50 100644
--- a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/data/partition/SerializedPartition.java
+++ b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/data/partition/SerializedPartition.java
@@ -48,7 +48,7 @@ public final class SerializedPartition<K> implements Partition<byte[], K> {
private static final Logger LOG = LoggerFactory.getLogger(SerializedPartition.class.getName());
private final K key;
- private volatile byte[] serializedData;
+ private volatile byte[] serializedData; // intentionally volatilize the reference
private volatile int length;
private volatile boolean committed;
// Will be null when the partition is committed when it is constructed.
diff --git a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/BlockInputReader.java b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/BlockInputReader.java
index b03b395..746a8f7 100644
--- a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/BlockInputReader.java
+++ b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/BlockInputReader.java
@@ -67,33 +67,37 @@ public final class BlockInputReader implements InputReader {
@Override
public List<CompletableFuture<DataUtil.IteratorWithNumBytes>> read() {
- final Optional<CommunicationPatternProperty.Value> comValue =
+ final Optional<CommunicationPatternProperty.Value> comValueOptional =
runtimeEdge.getPropertyValue(CommunicationPatternProperty.class);
-
- if (comValue.get().equals(CommunicationPatternProperty.Value.ONE_TO_ONE)) {
- return Collections.singletonList(readOneToOne());
- } else if (comValue.get().equals(CommunicationPatternProperty.Value.BROADCAST)) {
- return readBroadcast(index -> true);
- } else if (comValue.get().equals(CommunicationPatternProperty.Value.SHUFFLE)) {
- return readDataInRange(index -> true);
- } else {
- throw new UnsupportedCommPatternException(new Exception("Communication pattern not supported"));
+ final CommunicationPatternProperty.Value comValue = comValueOptional.orElseThrow(IllegalStateException::new);
+
+ switch (comValue) {
+ case ONE_TO_ONE:
+ return Collections.singletonList(readOneToOne());
+ case BROADCAST:
+ return readBroadcast(index -> true);
+ case SHUFFLE:
+ return readDataInRange(index -> true);
+ default:
+ throw new UnsupportedCommPatternException(new Exception("Communication pattern not supported"));
}
}
@Override
public CompletableFuture<DataUtil.IteratorWithNumBytes> retry(final int desiredIndex) {
- final Optional<CommunicationPatternProperty.Value> comValue =
+ final Optional<CommunicationPatternProperty.Value> comValueOptional =
runtimeEdge.getPropertyValue(CommunicationPatternProperty.class);
-
- if (comValue.get().equals(CommunicationPatternProperty.Value.ONE_TO_ONE)) {
- return readOneToOne();
- } else if (comValue.get().equals(CommunicationPatternProperty.Value.BROADCAST)) {
- return checkSingleElement(readBroadcast(index -> index == desiredIndex));
- } else if (comValue.get().equals(CommunicationPatternProperty.Value.SHUFFLE)) {
- return checkSingleElement(readDataInRange(index -> index == desiredIndex));
- } else {
- throw new UnsupportedCommPatternException(new Exception("Communication pattern not supported"));
+ final CommunicationPatternProperty.Value comValue = comValueOptional.orElseThrow(IllegalStateException::new);
+
+ switch (comValue) {
+ case ONE_TO_ONE:
+ return readOneToOne();
+ case BROADCAST:
+ return checkSingleElement(readBroadcast(index -> index == desiredIndex));
+ case SHUFFLE:
+ return checkSingleElement(readDataInRange(index -> index == desiredIndex));
+ default:
+ throw new UnsupportedCommPatternException(new Exception("Communication pattern not supported"));
}
}
diff --git a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/BlockOutputWriter.java b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/BlockOutputWriter.java
index 74d2909..97cde03 100644
--- a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/BlockOutputWriter.java
+++ b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/BlockOutputWriter.java
@@ -107,7 +107,7 @@ public final class BlockOutputWriter implements OutputWriter {
public void close() {
// Commit block.
final DataPersistenceProperty.Value persistence = (DataPersistenceProperty.Value) runtimeEdge
- .getPropertyValue(DataPersistenceProperty.class).get();
+ .getPropertyValue(DataPersistenceProperty.class).orElseThrow(IllegalStateException::new);
final Optional<Map<Integer, Long>> partitionSizeMap = blockToWrite.commit();
// Return the total size of the committed block.
diff --git a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/PipeInputReader.java b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/PipeInputReader.java
index 9af48d1..ac8efd4 100644
--- a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/PipeInputReader.java
+++ b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/PipeInputReader.java
@@ -59,13 +59,14 @@ public final class PipeInputReader implements InputReader {
@Override
public List<CompletableFuture<DataUtil.IteratorWithNumBytes>> read() {
- final Optional<CommunicationPatternProperty.Value> comValue =
+ final Optional<CommunicationPatternProperty.Value> comValueOptional =
runtimeEdge.getPropertyValue(CommunicationPatternProperty.class);
+ final CommunicationPatternProperty.Value comValue = comValueOptional.orElseThrow(IllegalStateException::new);
- if (comValue.get().equals(CommunicationPatternProperty.Value.ONE_TO_ONE)) {
+ if (comValue.equals(CommunicationPatternProperty.Value.ONE_TO_ONE)) {
return Collections.singletonList(pipeManagerWorker.read(dstTaskIndex, runtimeEdge, dstTaskIndex));
- } else if (comValue.get().equals(CommunicationPatternProperty.Value.BROADCAST)
- || comValue.get().equals(CommunicationPatternProperty.Value.SHUFFLE)) {
+ } else if (comValue.equals(CommunicationPatternProperty.Value.BROADCAST)
+ || comValue.equals(CommunicationPatternProperty.Value.SHUFFLE)) {
final int numSrcTasks = InputReader.getSourceParallelism(this);
final List<CompletableFuture<DataUtil.IteratorWithNumBytes>> futures = new ArrayList<>();
for (int srcTaskIdx = 0; srcTaskIdx < numSrcTasks; srcTaskIdx++) {
diff --git a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/PipeOutputWriter.java b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/PipeOutputWriter.java
index 540a984..277a135 100644
--- a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/PipeOutputWriter.java
+++ b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/PipeOutputWriter.java
@@ -137,14 +137,17 @@ public final class PipeOutputWriter implements OutputWriter {
}
private List<ByteOutputContext> getPipeToWrite(final Object element) {
- final CommunicationPatternProperty.Value comm =
- (CommunicationPatternProperty.Value) runtimeEdge.getPropertyValue(CommunicationPatternProperty.class).get();
- if (comm.equals(CommunicationPatternProperty.Value.ONE_TO_ONE)) {
- return Collections.singletonList(pipes.get(0));
- } else if (comm.equals(CommunicationPatternProperty.Value.BROADCAST)) {
- return pipes;
- } else {
- return Collections.singletonList(pipes.get((int) partitioner.partition(element)));
+ final Optional<CommunicationPatternProperty.Value> comValueOptional =
+ runtimeEdge.getPropertyValue(CommunicationPatternProperty.class);
+ final CommunicationPatternProperty.Value comm = comValueOptional.orElseThrow(IllegalStateException::new);
+
+ switch (comm) {
+ case ONE_TO_ONE:
+ return Collections.singletonList(pipes.get(0));
+ case BROADCAST:
+ return pipes;
+ default:
+ return Collections.singletonList(pipes.get((int) partitioner.partition(element)));
}
}
}
diff --git a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/task/MultiThreadParentTaskDataFetcher.java b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/task/MultiThreadParentTaskDataFetcher.java
index 6bde9a2..4cdea10 100644
--- a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/task/MultiThreadParentTaskDataFetcher.java
+++ b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/task/MultiThreadParentTaskDataFetcher.java
@@ -136,7 +136,7 @@ class MultiThreadParentTaskDataFetcher extends DataFetcher {
countBytesSynchronized(iterator);
elementQueue.offer(Finishmark.getInstance());
} else {
- exception.printStackTrace();
+ LOG.error(exception.getMessage());
throw new RuntimeException(exception);
}
});
diff --git a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/task/TaskExecutor.java b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/task/TaskExecutor.java
index 18a02de..7211e50 100644
--- a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/task/TaskExecutor.java
+++ b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/task/TaskExecutor.java
@@ -501,6 +501,7 @@ public final class TaskExecutor {
try {
Thread.sleep(pollingInterval);
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
e.printStackTrace();
throw new RuntimeException(e);
}
diff --git a/runtime/master/src/main/java/org/apache/nemo/runtime/master/BlockMetadata.java b/runtime/master/src/main/java/org/apache/nemo/runtime/master/BlockMetadata.java
index 11353a3..782f67d 100644
--- a/runtime/master/src/main/java/org/apache/nemo/runtime/master/BlockMetadata.java
+++ b/runtime/master/src/main/java/org/apache/nemo/runtime/master/BlockMetadata.java
@@ -36,7 +36,7 @@ final class BlockMetadata {
private static final Logger LOG = LoggerFactory.getLogger(BlockMetadata.class.getName());
private final String blockId;
private final BlockState blockState;
- private volatile BlockManagerMaster.BlockRequestHandler locationHandler;
+ private final BlockManagerMaster.BlockRequestHandler locationHandler;
/**
* Constructs the metadata for a block.
diff --git a/runtime/master/src/main/java/org/apache/nemo/runtime/master/PipeManagerMaster.java b/runtime/master/src/main/java/org/apache/nemo/runtime/master/PipeManagerMaster.java
index 6721d97..fc6145e 100644
--- a/runtime/master/src/main/java/org/apache/nemo/runtime/master/PipeManagerMaster.java
+++ b/runtime/master/src/main/java/org/apache/nemo/runtime/master/PipeManagerMaster.java
@@ -143,6 +143,7 @@ public final class PipeManagerMaster {
.build())
.build());
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw new RuntimeException(e);
} finally {
lock.unlock();
diff --git a/runtime/master/src/main/java/org/apache/nemo/runtime/master/resource/ContainerManager.java b/runtime/master/src/main/java/org/apache/nemo/runtime/master/resource/ContainerManager.java
index e1c21dc..8ec14a6 100644
--- a/runtime/master/src/main/java/org/apache/nemo/runtime/master/resource/ContainerManager.java
+++ b/runtime/master/src/main/java/org/apache/nemo/runtime/master/resource/ContainerManager.java
@@ -186,7 +186,10 @@ public final class ContainerManager {
try {
messageSender =
messageEnvironment.asyncConnect(executorId, MessageEnvironment.EXECUTOR_MESSAGE_LISTENER_ID).get();
- } catch (final InterruptedException | ExecutionException e) {
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ messageSender = new FailedMessageSender();
+ } catch (final ExecutionException e) {
// TODO #140: Properly classify and handle each RPC failure
messageSender = new FailedMessageSender();
}