You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by sb...@apache.org on 2015/03/27 16:37:39 UTC
[2/3] incubator-streams git commit: switch to use uninterruptibles
for thread sleep,
and add additional pausing to tasks experiencing race conditions on
interrupts.
switch to use uninterruptibles for thread sleep, and add additional pausing to tasks experiencing race conditions on interrupts.
this is not the ideal fix to the root cause which I believe to be sloppy thread management throughout streams-runtime-local, but it is sufficient to ensure the integration test proposed for incubator-streams-examples/elasticsearch-hdfs passes.
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/75846d9a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/75846d9a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/75846d9a
Branch: refs/heads/master
Commit: 75846d9a26d0f88ae3cac42e09e98ca0a4fbbaf5
Parents: ed242de
Author: Steve Blackmon (@steveblackmon) <sb...@apache.org>
Authored: Mon Mar 23 13:03:36 2015 -0500
Committer: Steve Blackmon (@steveblackmon) <sb...@apache.org>
Committed: Mon Mar 23 13:03:36 2015 -0500
----------------------------------------------------------------------
.../ElasticsearchPersistWriter.java | 6 +++---
.../streams/hdfs/WebHdfsPersistReaderTask.java | 3 +++
.../local/builders/LocalStreamBuilder.java | 12 +++++++-----
.../local/tasks/StreamsPersistWriterTask.java | 15 ++++-----------
.../local/tasks/StreamsProviderTask.java | 20 +++++++-------------
5 files changed, 24 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/75846d9a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java
index 3179a69..c984a12 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java
@@ -22,6 +22,7 @@ package org.apache.streams.elasticsearch;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.streams.config.StreamsConfigurator;
import org.apache.streams.core.*;
import org.apache.streams.jackson.StreamsJacksonMapper;
@@ -45,6 +46,7 @@ import java.io.Serializable;
import java.text.DecimalFormat;
import java.text.NumberFormat;
import java.util.*;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -173,6 +175,7 @@ public class ElasticsearchPersistWriter implements StreamsPersistWriter, DatumSt
}
public void cleanUp() {
+
try {
LOGGER.debug("cleanUp started");
@@ -180,9 +183,6 @@ public class ElasticsearchPersistWriter implements StreamsPersistWriter, DatumSt
// before they close, check to ensure that
flushInternal();
- // before they close, check to ensure that
- flushInternal();
-
LOGGER.debug("flushInternal completed");
waitToCatchUp(0, 5 * 60 * 1000);
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/75846d9a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReaderTask.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReaderTask.java b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReaderTask.java
index f3877e3..4d1c43d 100644
--- a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReaderTask.java
+++ b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReaderTask.java
@@ -21,6 +21,7 @@ package org.apache.streams.hdfs;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.base.Strings;
import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.hadoop.fs.FileStatus;
import org.apache.streams.core.DatumStatus;
import org.apache.streams.core.StreamsDatum;
@@ -33,6 +34,7 @@ import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
public class WebHdfsPersistReaderTask implements Runnable {
@@ -91,6 +93,7 @@ public class WebHdfsPersistReaderTask implements Runnable {
}
}
+ Uninterruptibles.sleepUninterruptibly(15, TimeUnit.SECONDS);
}
private void write( StreamsDatum entry ) {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/75846d9a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java
index ae519e1..556b79c 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java
@@ -18,6 +18,7 @@
package org.apache.streams.local.builders;
+import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.streams.core.*;
import org.apache.streams.local.counters.StreamsTaskCounter;
import org.apache.streams.local.executors.ShutdownStreamOnUnhandleThrowableThreadPoolExecutor;
@@ -228,16 +229,17 @@ public class LocalStreamBuilder implements StreamBuilder {
isRunning = isRunning || (tasksRunning && task.getInBoundQueue().size() > 0);
}
if(isRunning) {
- Thread.sleep(3000);
- } else {
- LOGGER.info("Stream has completed successfully, shutting down @ {}", System.currentTimeMillis());
+ Uninterruptibles.sleepUninterruptibly(2500, TimeUnit.MILLISECONDS);
}
+ LOGGER.info("Stream has completed successfully, pausing @ {}", System.currentTimeMillis());
}
LOGGER.debug("Components are no longer running or timed out");
- } catch (InterruptedException e){
- LOGGER.warn("Runtime interrupted. Beginning shutdown");
+ } catch (Exception e){
+ LOGGER.warn("Runtime exception. Beginning shutdown");
forcedShutDown = true;
} finally{
+ Uninterruptibles.sleepUninterruptibly(5000, TimeUnit.MILLISECONDS);
+ LOGGER.info("Stream has completed successfully, shutting down @ {}", System.currentTimeMillis());
stopInternal(forcedShutDown);
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/75846d9a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsPersistWriterTask.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsPersistWriterTask.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsPersistWriterTask.java
index 0bd4af9..860a8f0 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsPersistWriterTask.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsPersistWriterTask.java
@@ -18,6 +18,7 @@
package org.apache.streams.local.tasks;
+import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.streams.core.*;
import org.apache.streams.core.util.DatumUtils;
import org.apache.streams.local.counters.StreamsTaskCounter;
@@ -37,7 +38,7 @@ public class StreamsPersistWriterTask extends BaseStreamsTask implements DatumSt
private final static Logger LOGGER = LoggerFactory.getLogger(StreamsPersistWriterTask.class);
private StreamsPersistWriter writer;
- private long sleepTime;
+ private long sleepTime = DEFAULT_SLEEP_TIME_MS * 10;
private AtomicBoolean keepRunning;
private Map<String, Object> streamConfig;
private BlockingQueue<StreamsDatum> inQueue;
@@ -139,19 +140,11 @@ public class StreamsPersistWriterTask extends BaseStreamsTask implements DatumSt
LOGGER.debug("Received null StreamsDatum @ writer : {}", this.writer.getClass().getName());
}
}
- try {
- Thread.sleep(sleepTime);
- } catch (InterruptedException e) {
- LOGGER.error("Sleep interrupted: {}", e);
- }
+ Uninterruptibles.sleepUninterruptibly(sleepTime, TimeUnit.MILLISECONDS);
} catch(Exception e) {
LOGGER.error("Failed to execute Persist Writer {}",this.writer.getClass().getSimpleName(), e);
} finally {
- try {
- Thread.sleep(sleepTime);
- } catch (InterruptedException e) {
- LOGGER.error("Sleep interrupted: {}", e);
- }
+ Uninterruptibles.sleepUninterruptibly(sleepTime, TimeUnit.MILLISECONDS);
this.writer.cleanUp();
this.isRunning.set(false);
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/75846d9a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProviderTask.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProviderTask.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProviderTask.java
index 6d48490..c446a72 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProviderTask.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProviderTask.java
@@ -18,6 +18,7 @@
package org.apache.streams.local.tasks;
+import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.streams.core.*;
import org.apache.streams.core.util.DatumUtils;
import org.apache.streams.local.counters.StreamsTaskCounter;
@@ -30,6 +31,7 @@ import java.util.Map;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
@@ -173,18 +175,14 @@ public class StreamsProviderTask extends BaseStreamsTask implements DatumStatusC
if(zeros > maxZeros)
this.keepRunning.set(false);
if(zeros > 0)
- Thread.sleep(sleepTime);
- } catch (InterruptedException e) {
+ Uninterruptibles.sleepUninterruptibly(sleepTime, TimeUnit.MILLISECONDS);
+ } catch (Exception e) {
this.counter.incrementErrorCount();
- LOGGER.warn("Thread interrupted");
+ LOGGER.warn("Thread exception");
this.keepRunning.set(false);
}
}
- try {
- Thread.sleep(sleepTime);
- } catch (InterruptedException e) {
- LOGGER.error("Sleep interrupted: {}", e);
- }
+ Uninterruptibles.sleepUninterruptibly(sleepTime, TimeUnit.MILLISECONDS);
}
break;
case READ_CURRENT:
@@ -207,11 +205,7 @@ public class StreamsProviderTask extends BaseStreamsTask implements DatumStatusC
} catch( Exception e ) {
LOGGER.error("Error in processing provider stream", e);
} finally {
- try {
- Thread.sleep(sleepTime);
- } catch (InterruptedException e) {
- LOGGER.error("Sleep interrupted: {}", e);
- }
+ Uninterruptibles.sleepUninterruptibly(sleepTime, TimeUnit.MILLISECONDS);
LOGGER.debug("Complete Provider Task execution for {}", this.provider.getClass().getSimpleName());
this.provider.cleanUp();
//Setting started to 'true' here will allow the isRunning() method to return false in the event of an exception