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/08/28 22:16:08 UTC
incubator-streams git commit: resolves STREAMS-357 now we can keep a
stream alive for a minimum duration and change how often to check for a
shutdown condition
Repository: incubator-streams
Updated Branches:
refs/heads/STREAMS-357 [created] 2dad128cd
resolves STREAMS-357
now we can keep a stream alive for a minimum duration and change how often to check for a shutdown condition
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/2dad128c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/2dad128c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/2dad128c
Branch: refs/heads/STREAMS-357
Commit: 2dad128cdcd206149cf7199e857e89f4956a2d84
Parents: e52cb25
Author: Steve Blackmon (@steveblackmon) <sb...@apache.org>
Authored: Fri Aug 28 15:16:06 2015 -0500
Committer: Steve Blackmon (@steveblackmon) <sb...@apache.org>
Committed: Fri Aug 28 15:16:06 2015 -0500
----------------------------------------------------------------------
.../src/main/jsonschema/StreamsConfiguration.json | 10 ++++++++++
.../streams/monitoring/tasks/BroadcastMonitorThread.java | 3 ++-
.../apache/streams/local/builders/LocalStreamBuilder.java | 3 ++-
.../streams/local/tasks/StreamsPersistWriterTask.java | 4 ++--
.../apache/streams/local/tasks/StreamsProcessorTask.java | 6 +++---
5 files changed, 19 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2dad128c/streams-config/src/main/jsonschema/StreamsConfiguration.json
----------------------------------------------------------------------
diff --git a/streams-config/src/main/jsonschema/StreamsConfiguration.json b/streams-config/src/main/jsonschema/StreamsConfiguration.json
index d331df2..ee8c37d 100644
--- a/streams-config/src/main/jsonschema/StreamsConfiguration.json
+++ b/streams-config/src/main/jsonschema/StreamsConfiguration.json
@@ -38,6 +38,16 @@
"description" : "Size of inter-component buffers",
"default": 1000
},
+ "shutdownCheckDelay": {
+ "type": "integer",
+ "description": "Initial delay before checking for shutdown",
+ "default": 1000
+ },
+ "shutdownCheckInterval": {
+ "type": "integer",
+ "description": "Recurring delay between checking for shutdown",
+ "default": 1000
+ },
"startedAt": {
"type": "integer",
"description": "Timestamp of beginning of stream execution",
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2dad128c/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java
----------------------------------------------------------------------
diff --git a/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java b/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java
index 20c605e..f21a212 100644
--- a/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java
+++ b/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java
@@ -160,7 +160,8 @@ public class BroadcastMonitorThread extends NotificationBroadcasterSupport imple
server = ManagementFactory.getPlatformMBeanServer();
- if (this.configuration != null) {
+ if (this.configuration != null &&
+ this.configuration.getBroadcastURI() != null) {
try {
broadcastURI = new URI(configuration.getBroadcastURI());
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2dad128c/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 f31c7ed..a7dfa37 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
@@ -257,6 +257,7 @@ public class LocalStreamBuilder implements StreamBuilder {
setupProviderTasks(provTasks);
LOGGER.info("Started stream with {} components", tasks.size());
while(isRunning) {
+ Uninterruptibles.sleepUninterruptibly(streamConfig.getShutdownCheckDelay(), TimeUnit.MILLISECONDS);
isRunning = false;
for(StreamsProviderTask task : provTasks.values()) {
isRunning = isRunning || task.isRunning();
@@ -271,7 +272,7 @@ public class LocalStreamBuilder implements StreamBuilder {
isRunning = isRunning || (tasksRunning && task.getInBoundQueue().size() > 0);
}
if(isRunning) {
- Uninterruptibles.sleepUninterruptibly(2500, TimeUnit.MILLISECONDS);
+ Uninterruptibles.sleepUninterruptibly(streamConfig.getShutdownCheckInterval(), TimeUnit.MILLISECONDS);
}
}
LOGGER.info("Components are no longer running or timed out");
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2dad128c/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 21954d4..fb97218 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
@@ -133,12 +133,12 @@ public class StreamsPersistWriterTask extends BaseStreamsTask implements DatumSt
this.counter.incrementErrorCount();
}
} else { //datums should never be null
- LOGGER.debug("Received null StreamsDatum @ writer : {}", this.writer.getClass().getName());
+ LOGGER.trace("Received null StreamsDatum @ writer : {}", this.writer.getClass().getName());
}
}
Uninterruptibles.sleepUninterruptibly(streamConfig.getBatchFrequencyMs(), TimeUnit.MILLISECONDS);
} catch(Throwable e) {
- LOGGER.error("Caught Throwable in Persist Writer {}", this.writer.getClass().getSimpleName(), e);
+ LOGGER.error("Caught Throwable in Persist Writer {} : {}", this.writer.getClass().getSimpleName(), e);
} finally {
Uninterruptibles.sleepUninterruptibly(streamConfig.getBatchFrequencyMs(), TimeUnit.MILLISECONDS);
this.writer.cleanUp();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2dad128c/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProcessorTask.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProcessorTask.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProcessorTask.java
index 629226c..137c7e1 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProcessorTask.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProcessorTask.java
@@ -136,18 +136,18 @@ public class StreamsProcessorTask extends BaseStreamsTask implements DatumStatus
}
}
} catch (InterruptedException ie) {
- LOGGER.warn("Received InteruptedException, shutting down and re-applying interrupt status.");
+ LOGGER.warn("Received InterruptedException, shutting down and re-applying interrupt status.");
this.keepRunning.set(false);
Thread.currentThread().interrupt();
} catch (Throwable t) {
this.counter.incrementErrorCount();
- LOGGER.warn("Caught Throwable in processor, {} : {}", this.processor.getClass().getName(), t.getMessage());
+ LOGGER.warn("Caught Throwable in processor, {} : {}", this.processor.getClass().getName(), t);
statusCounter.incrementStatus(DatumStatus.FAIL);
//Add the error to the metadata, but keep processing
DatumUtils.addErrorToMetadata(datum, t, this.processor.getClass());
}
} else {
- LOGGER.debug("Removed NULL datum from queue at processor : {}", this.processor.getClass().getName());
+ LOGGER.trace("Removed NULL datum from queue at processor : {}", this.processor.getClass().getName());
}
}
} catch(Throwable e) {