You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by re...@apache.org on 2014/10/20 19:28:53 UTC
[01/11] git commit: Implement ThroughputQueues and removed busy
waiting on processors and writer task
Repository: incubator-streams
Updated Branches:
refs/heads/master 0a32159d8 -> 7b01eb48b
Implement ThroughputQueues and removed busy waiting on processors and writer task
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/fb11add0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/fb11add0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/fb11add0
Branch: refs/heads/master
Commit: fb11add0e432e3ba63322dfef863783c5033f0f8
Parents: 50906ed
Author: Ryan Ebanks <ry...@gmail.com>
Authored: Mon Oct 6 17:10:41 2014 -0500
Committer: Ryan Ebanks <ry...@gmail.com>
Committed: Mon Oct 6 17:10:41 2014 -0500
----------------------------------------------------------------------
.../local/builders/LocalStreamBuilder.java | 31 +++++----
.../streams/local/builders/StreamComponent.java | 15 +++--
.../streams/local/tasks/BaseStreamsTask.java | 45 +++++++++----
.../local/tasks/StreamsPersistWriterTask.java | 55 +++++++++++-----
.../local/tasks/StreamsProcessorTask.java | 68 ++++++++++++++------
.../local/tasks/StreamsProviderTask.java | 3 +-
.../apache/streams/local/tasks/StreamsTask.java | 9 +--
7 files changed, 152 insertions(+), 74 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/fb11add0/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 bec1ff9..0b02f4e 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
@@ -21,17 +21,16 @@ package org.apache.streams.local.builders;
import org.apache.log4j.spi.LoggerFactory;
import org.apache.streams.core.*;
import org.apache.streams.local.executors.ShutdownStreamOnUnhandleThrowableThreadPoolExecutor;
+import org.apache.streams.local.queues.ThroughputQueue;
import org.apache.streams.local.tasks.LocalStreamProcessMonitorThread;
import org.apache.streams.local.tasks.StatusCounterMonitorThread;
import org.apache.streams.local.tasks.StreamsProviderTask;
import org.apache.streams.local.tasks.StreamsTask;
-import org.apache.streams.util.SerializationUtil;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import java.math.BigInteger;
import java.util.*;
-import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -48,7 +47,6 @@ public class LocalStreamBuilder implements StreamBuilder {
public static final String TIMEOUT_KEY = "TIMEOUT";
private Map<String, StreamComponent> providers;
private Map<String, StreamComponent> components;
- private Queue<StreamsDatum> queue;
private Map<String, Object> streamConfig;
private ExecutorService executor;
private ExecutorService monitor;
@@ -57,12 +55,13 @@ public class LocalStreamBuilder implements StreamBuilder {
private LocalStreamProcessMonitorThread monitorThread;
private Map<String, List<StreamsTask>> tasks;
private Thread shutdownHook;
+ private int maxQueueCapacity;
/**
*
*/
public LocalStreamBuilder(){
- this(new ConcurrentLinkedQueue<StreamsDatum>(), null);
+ this(-1, null);
}
/**
@@ -70,29 +69,29 @@ public class LocalStreamBuilder implements StreamBuilder {
* @param streamConfig
*/
public LocalStreamBuilder(Map<String, Object> streamConfig) {
- this(new ConcurrentLinkedQueue<StreamsDatum>(), streamConfig);
+ this(-1, streamConfig);
}
/**
*
- * @param queueType
+ * @param maxQueueCapacity
*/
- public LocalStreamBuilder(Queue<StreamsDatum> queueType) {
- this(queueType, null);
+ public LocalStreamBuilder(int maxQueueCapacity) {
+ this(maxQueueCapacity, null);
}
/**
*
- * @param queueType
+ * @param maxQueueCapacity
* @param streamConfig
*/
- public LocalStreamBuilder(Queue<StreamsDatum> queueType, Map<String, Object> streamConfig) {
- this.queue = queueType;
+ public LocalStreamBuilder(int maxQueueCapacity, Map<String, Object> streamConfig) {
this.providers = new HashMap<String, StreamComponent>();
this.components = new HashMap<String, StreamComponent>();
this.streamConfig = streamConfig;
this.totalTasks = 0;
this.monitorTasks = 0;
+ this.maxQueueCapacity = maxQueueCapacity;
final LocalStreamBuilder self = this;
this.shutdownHook = new Thread() {
@Override
@@ -146,7 +145,7 @@ public class LocalStreamBuilder implements StreamBuilder {
@Override
public StreamBuilder addStreamsProcessor(String id, StreamsProcessor processor, int numTasks, String... inBoundIds) {
validateId(id);
- StreamComponent comp = new StreamComponent(id, processor, cloneQueue(), numTasks);
+ StreamComponent comp = new StreamComponent(id, processor, new ThroughputQueue<StreamsDatum>(this.maxQueueCapacity, id), numTasks);
this.components.put(id, comp);
connectToOtherComponents(inBoundIds, comp);
this.totalTasks += numTasks;
@@ -158,7 +157,7 @@ public class LocalStreamBuilder implements StreamBuilder {
@Override
public StreamBuilder addStreamsPersistWriter(String id, StreamsPersistWriter writer, int numTasks, String... inBoundIds) {
validateId(id);
- StreamComponent comp = new StreamComponent(id, writer, cloneQueue(), numTasks);
+ StreamComponent comp = new StreamComponent(id, writer, new ThroughputQueue<StreamsDatum>(this.maxQueueCapacity, id), numTasks);
this.components.put(id, comp);
connectToOtherComponents(inBoundIds, comp);
this.totalTasks += numTasks;
@@ -373,13 +372,13 @@ public class LocalStreamBuilder implements StreamBuilder {
private void validateId(String id) {
if(this.providers.containsKey(id) || this.components.containsKey(id)) {
throw new InvalidStreamException("Duplicate id. "+id+" is already assigned to another component");
+ } else if(id.contains(":")) {
+ throw new InvalidStreamException("Invalid character, ':', in component id : "+id);
}
}
- private Queue<StreamsDatum> cloneQueue() {
- return (Queue<StreamsDatum>) SerializationUtil.cloneBySerialization(this.queue);
- }
+
protected int getTimeout() {
//Set the timeout of it is configured, otherwise signal downstream components to use their default
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/fb11add0/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java
index 8611bbb..5131227 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java
@@ -28,6 +28,7 @@ import org.joda.time.DateTime;
import java.math.BigInteger;
import java.util.*;
+import java.util.concurrent.BlockingQueue;
/**
* Stores the implementations of {@link org.apache.streams.core.StreamsOperation}, the StreamsOperations it is connected
@@ -40,8 +41,8 @@ public class StreamComponent {
private String id;
private Set<StreamComponent> inBound;
- private Map<StreamComponent, Queue<StreamsDatum>> outBound;
- private Queue<StreamsDatum> inQueue;
+ private Map<StreamComponent, BlockingQueue<StreamsDatum>> outBound;
+ private BlockingQueue<StreamsDatum> inQueue;
private StreamsProvider provider;
private StreamsProcessor processor;
private StreamsPersistWriter writer;
@@ -98,7 +99,7 @@ public class StreamComponent {
* @param inQueue
* @param numTasks
*/
- public StreamComponent(String id, StreamsProcessor processor, Queue<StreamsDatum> inQueue, int numTasks) {
+ public StreamComponent(String id, StreamsProcessor processor, BlockingQueue<StreamsDatum> inQueue, int numTasks) {
this.id = id;
this.processor = processor;
this.inQueue = inQueue;
@@ -113,7 +114,7 @@ public class StreamComponent {
* @param inQueue
* @param numTasks
*/
- public StreamComponent(String id, StreamsPersistWriter writer, Queue<StreamsDatum> inQueue, int numTasks) {
+ public StreamComponent(String id, StreamsPersistWriter writer, BlockingQueue<StreamsDatum> inQueue, int numTasks) {
this.id = id;
this.writer = writer;
this.inQueue = inQueue;
@@ -123,7 +124,7 @@ public class StreamComponent {
private void initializePrivateVariables() {
this.inBound = new HashSet<StreamComponent>();
- this.outBound = new HashMap<StreamComponent, Queue<StreamsDatum>>();
+ this.outBound = new HashMap<StreamComponent, BlockingQueue<StreamsDatum>>();
}
/**
@@ -131,7 +132,7 @@ public class StreamComponent {
* @param component the component that this supplying their inbound queue
* @param queue the queue to to put post processed/provided datums on
*/
- public void addOutBoundQueue(StreamComponent component, Queue<StreamsDatum> queue) {
+ public void addOutBoundQueue(StreamComponent component, BlockingQueue<StreamsDatum> queue) {
this.outBound.put(component, queue);
}
@@ -163,7 +164,7 @@ public class StreamComponent {
* The inbound queue for this component
* @return inbound queue
*/
- public Queue<StreamsDatum> getInBoundQueue() {
+ public BlockingQueue<StreamsDatum> getInBoundQueue() {
return this.inQueue;
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/fb11add0/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java
index 980d199..758a883 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java
@@ -20,6 +20,7 @@ package org.apache.streams.local.tasks;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.Lists;
import org.apache.streams.core.StreamsDatum;
import org.apache.streams.jackson.StreamsJacksonMapper;
import org.apache.streams.pojo.json.Activity;
@@ -30,6 +31,8 @@ import org.slf4j.LoggerFactory;
import java.io.Serializable;
import java.util.*;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
/**
*
@@ -38,8 +41,8 @@ public abstract class BaseStreamsTask implements StreamsTask {
private static final Logger LOGGER = LoggerFactory.getLogger(BaseStreamsTask.class);
- private List<Queue<StreamsDatum>> inQueues = new ArrayList<Queue<StreamsDatum>>();
- private List<Queue<StreamsDatum>> outQueues = new LinkedList<Queue<StreamsDatum>>();
+ private List<BlockingQueue<StreamsDatum>> inQueues = new ArrayList<BlockingQueue<StreamsDatum>>();
+ private List<BlockingQueue<StreamsDatum>> outQueues = new LinkedList<BlockingQueue<StreamsDatum>>();
private int inIndex = 0;
private ObjectMapper mapper;
@@ -50,31 +53,31 @@ public abstract class BaseStreamsTask implements StreamsTask {
@Override
- public void addInputQueue(Queue<StreamsDatum> inputQueue) {
+ public void addInputQueue(BlockingQueue<StreamsDatum> inputQueue) {
this.inQueues.add(inputQueue);
}
@Override
- public void addOutputQueue(Queue<StreamsDatum> outputQueue) {
+ public void addOutputQueue(BlockingQueue<StreamsDatum> outputQueue) {
this.outQueues.add(outputQueue);
}
@Override
- public List<Queue<StreamsDatum>> getInputQueues() {
+ public List<BlockingQueue<StreamsDatum>> getInputQueues() {
return this.inQueues;
}
@Override
- public List<Queue<StreamsDatum>> getOutputQueues() {
+ public List<BlockingQueue<StreamsDatum>> getOutputQueues() {
return this.outQueues;
}
/**
- * NOTE NECESSARY AT THE MOMENT. MAY BECOME NECESSARY AS WE LOOK AT MAKING JOIN TASKS. CURRENTLY ALL TASK HAVE MAX
- * OF 1 INPUT QUEUE.
+ * SHOULD NOT BE NECCESARY, WILL REMOVE.
* Round Robins through input queues to get the next StreamsDatum. If all input queues are empty, it will return null.
* @return the next StreamsDatum or null if all input queues are empty.
*/
+ @Deprecated
protected StreamsDatum getNextDatum() {
int startIndex = this.inIndex;
int index = startIndex;
@@ -91,23 +94,41 @@ public abstract class BaseStreamsTask implements StreamsTask {
* clones of the datum and adds a new clone to each queue.
* @param datum
*/
- protected void addToOutgoingQueue(StreamsDatum datum) {
+ protected void addToOutgoingQueue(StreamsDatum datum) throws InterruptedException{
if(this.outQueues.size() == 1) {
- ComponentUtils.offerUntilSuccess(datum, outQueues.get(0));
+ outQueues.get(0).put(datum);
}
else {
StreamsDatum newDatum = null;
- for(Queue<StreamsDatum> queue : this.outQueues) {
+ List<BlockingQueue<StreamsDatum>> failedQueues = Lists.newLinkedList();
+ // TODO
+ // Needs to be optimized better but workable now
+ // Adds datums to queues that aren't full, then adds to full queues with blocking
+ for(BlockingQueue<StreamsDatum> queue : this.outQueues) {
try {
newDatum = cloneStreamsDatum(datum);
if(newDatum != null) {
- ComponentUtils.offerUntilSuccess(newDatum, queue);
+ if(!queue.offer(newDatum, 500, TimeUnit.MILLISECONDS)) {
+ failedQueues.add(queue);
+ }
}
} catch (RuntimeException e) {
LOGGER.debug("Failed to add StreamsDatum to outgoing queue : {}", datum);
LOGGER.error("Exception while offering StreamsDatum to outgoing queue: {}", e);
}
}
+ for(BlockingQueue<StreamsDatum> queue : failedQueues) {
+ try {
+ newDatum = cloneStreamsDatum(datum);
+ if(newDatum != null) {
+ queue.put(newDatum);
+ }
+ } catch (RuntimeException e) {
+ LOGGER.debug("Failed to add StreamsDatum to outgoing queue : {}", datum);
+ LOGGER.error("Exception while offering StreamsDatum to outgoing queue: {}", e);
+ }
+ }
+
}
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/fb11add0/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 59c438e..ce9f5a7 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
@@ -27,6 +27,7 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
/**
@@ -40,7 +41,7 @@ public class StreamsPersistWriterTask extends BaseStreamsTask implements DatumSt
private long sleepTime;
private AtomicBoolean keepRunning;
private Map<String, Object> streamConfig;
- private Queue<StreamsDatum> inQueue;
+ private BlockingQueue<StreamsDatum> inQueue;
private AtomicBoolean isRunning;
private DatumStatusCounter statusCounter = new DatumStatusCounter();
@@ -77,7 +78,7 @@ public class StreamsPersistWriterTask extends BaseStreamsTask implements DatumSt
}
@Override
- public void addInputQueue(Queue<StreamsDatum> inputQueue) {
+ public void addInputQueue(BlockingQueue<StreamsDatum> inputQueue) {
this.inQueue = inputQueue;
}
@@ -90,29 +91,51 @@ public class StreamsPersistWriterTask extends BaseStreamsTask implements DatumSt
public void run() {
try {
this.writer.prepare(this.streamConfig);
- StreamsDatum datum = this.inQueue.poll();
while(this.keepRunning.get()) {
+ StreamsDatum datum = null;
+ try {
+ datum = this.inQueue.take();
+ } catch (InterruptedException ie) {
+ LOGGER.error("Received InterruptedException. Shutting down and re-applying interrupt status.");
+ Thread.currentThread().interrupt();
+ }
if(datum != null) {
try {
this.writer.write(datum);
statusCounter.incrementStatus(DatumStatus.SUCCESS);
} catch (Exception e) {
LOGGER.error("Error writing to persist writer {}", this.writer.getClass().getSimpleName(), e);
- this.keepRunning.set(false);
+ this.keepRunning.set(false); // why do we shutdown on a failed write ?
statusCounter.incrementStatus(DatumStatus.FAIL);
DatumUtils.addErrorToMetadata(datum, e, this.writer.getClass());
}
+ } else { //datums should never be null
+ LOGGER.warn("Received null StreamsDatum @ writer : {}", this.writer.getClass().getName());
}
- else {
- try {
- Thread.sleep(this.sleepTime);
- } catch (InterruptedException e) {
- LOGGER.warn("Thread interrupted in Writer task for {}",this.writer.getClass().getSimpleName(), e);
- this.keepRunning.set(false);
- }
- }
- datum = this.inQueue.poll();
}
+// StreamsDatum datum = this.inQueue.poll();
+// while(this.keepRunning.get()) {
+// if(datum != null) {
+// try {
+// this.writer.write(datum);
+// statusCounter.incrementStatus(DatumStatus.SUCCESS);
+// } catch (Exception e) {
+// LOGGER.error("Error writing to persist writer {}", this.writer.getClass().getSimpleName(), e);
+// this.keepRunning.set(false);
+// statusCounter.incrementStatus(DatumStatus.FAIL);
+// DatumUtils.addErrorToMetadata(datum, e, this.writer.getClass());
+// }
+// }
+// else {
+// try {
+// Thread.sleep(this.sleepTime);
+// } catch (InterruptedException e) {
+// LOGGER.warn("Thread interrupted in Writer task for {}",this.writer.getClass().getSimpleName(), e);
+// this.keepRunning.set(false);
+// }
+// }
+// datum = this.inQueue.poll();
+// }
} catch(Exception e) {
LOGGER.error("Failed to execute Persist Writer {}",this.writer.getClass().getSimpleName(), e);
@@ -129,13 +152,13 @@ public class StreamsPersistWriterTask extends BaseStreamsTask implements DatumSt
@Override
- public void addOutputQueue(Queue<StreamsDatum> outputQueue) {
+ public void addOutputQueue(BlockingQueue<StreamsDatum> outputQueue) {
throw new UnsupportedOperationException(this.getClass().getName()+" does not support method - setOutputQueue()");
}
@Override
- public List<Queue<StreamsDatum>> getInputQueues() {
- List<Queue<StreamsDatum>> queues = new LinkedList<Queue<StreamsDatum>>();
+ public List<BlockingQueue<StreamsDatum>> getInputQueues() {
+ List<BlockingQueue<StreamsDatum>> queues = new LinkedList<>();
queues.add(this.inQueue);
return queues;
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/fb11add0/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 beec0c2..4b65066 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
@@ -28,6 +28,7 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
/**
@@ -42,7 +43,7 @@ public class StreamsProcessorTask extends BaseStreamsTask implements DatumStatus
private long sleepTime;
private AtomicBoolean keepRunning;
private Map<String, Object> streamConfig;
- private Queue<StreamsDatum> inQueue;
+ private BlockingQueue<StreamsDatum> inQueue;
private AtomicBoolean isRunning;
private DatumStatusCounter statusCounter = new DatumStatusCounter();
@@ -83,7 +84,7 @@ public class StreamsProcessorTask extends BaseStreamsTask implements DatumStatus
}
@Override
- public void addInputQueue(Queue<StreamsDatum> inputQueue) {
+ public void addInputQueue(BlockingQueue<StreamsDatum> inputQueue) {
this.inQueue = inputQueue;
}
@@ -96,35 +97,66 @@ public class StreamsProcessorTask extends BaseStreamsTask implements DatumStatus
public void run() {
try {
this.processor.prepare(this.streamConfig);
- StreamsDatum datum = this.inQueue.poll();
while(this.keepRunning.get()) {
+ StreamsDatum datum = null;
+ try {
+ datum = this.inQueue.take();
+ } catch (InterruptedException ie) {
+ LOGGER.warn("Received InteruptedException, shutting down and re-applying interrupt status.");
+ Thread.currentThread().interrupt();
+ }
if(datum != null) {
try {
List<StreamsDatum> output = this.processor.process(datum);
-
if(output != null) {
for(StreamsDatum outDatum : output) {
- super.addToOutgoingQueue(outDatum);
+ super.addToOutgoingQueue(datum);
statusCounter.incrementStatus(DatumStatus.SUCCESS);
}
}
- } catch (Throwable e) {
- LOGGER.error("Throwable Streams Processor {}", e);
+ } catch (InterruptedException ie) {
+ LOGGER.warn("Received InteruptedException, shutting down and re-applying interrupt status.");
+ Thread.currentThread().interrupt();
+ } catch (Throwable t) {
+ LOGGER.warn("Caught Throwable in processor, {} : {}", this.processor.getClass().getName(), t.getMessage());
statusCounter.incrementStatus(DatumStatus.FAIL);
//Add the error to the metadata, but keep processing
- DatumUtils.addErrorToMetadata(datum, e, this.processor.getClass());
- }
- }
- else {
- try {
- Thread.sleep(this.sleepTime);
- } catch (InterruptedException e) {
- this.keepRunning.set(false);
+ DatumUtils.addErrorToMetadata(datum, t, this.processor.getClass());
}
+ } else {
+ LOGGER.warn("Removed NULL datum from queue at processor : {}", this.processor.getClass().getName());
}
- datum = this.inQueue.poll();
}
+// this.processor.prepare(this.streamConfig);
+// StreamsDatum datum = this.inQueue.poll();
+// while(this.keepRunning.get()) {
+// if(datum != null) {
+// try {
+// List<StreamsDatum> output = this.processor.process(datum);
+// if(output != null) {
+// for(StreamsDatum outDatum : output) {
+// super.addToOutgoingQueue(outDatum);
+// statusCounter.incrementStatus(DatumStatus.SUCCESS);
+// }
+// }
+// } catch (Throwable e) {
+// LOGGER.error("Throwable Streams Processor {}", e);
+// statusCounter.incrementStatus(DatumStatus.FAIL);
+// //Add the error to the metadata, but keep processing
+// DatumUtils.addErrorToMetadata(datum, e, this.processor.getClass());
+// }
+// }
+// else {
+// try {
+// Thread.sleep(this.sleepTime);
+// } catch (InterruptedException e) {
+// this.keepRunning.set(false);
+// }
+// }
+// datum = this.inQueue.poll();
+// }
+
} finally {
this.isRunning.set(false);
this.processor.cleanUp();
@@ -132,8 +164,8 @@ public class StreamsProcessorTask extends BaseStreamsTask implements DatumStatus
}
@Override
- public List<Queue<StreamsDatum>> getInputQueues() {
- List<Queue<StreamsDatum>> queues = new LinkedList<Queue<StreamsDatum>>();
+ public List<BlockingQueue<StreamsDatum>> getInputQueues() {
+ List<BlockingQueue<StreamsDatum>> queues = new LinkedList<BlockingQueue<StreamsDatum>>();
queues.add(this.inQueue);
return queues;
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/fb11add0/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 d040c4b..cea7f63 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
@@ -27,6 +27,7 @@ import org.slf4j.LoggerFactory;
import java.math.BigInteger;
import java.util.Map;
import java.util.Queue;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
/**
@@ -122,7 +123,7 @@ public class StreamsProviderTask extends BaseStreamsTask implements DatumStatusC
}
@Override
- public void addInputQueue(Queue<StreamsDatum> inputQueue) {
+ public void addInputQueue(BlockingQueue<StreamsDatum> inputQueue) {
throw new UnsupportedOperationException(this.getClass().getName()+" does not support method - setInputQueue()");
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/fb11add0/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsTask.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsTask.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsTask.java
index 1bb3ab1..45c25f3 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsTask.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsTask.java
@@ -23,6 +23,7 @@ import org.apache.streams.core.StreamsDatum;
import java.util.List;
import java.util.Map;
import java.util.Queue;
+import java.util.concurrent.BlockingQueue;
/**
* Interface for all task that will be used to execute instances of {@link org.apache.streams.core.StreamsOperation}
@@ -49,13 +50,13 @@ public interface StreamsTask extends Runnable{
* Add an input {@link java.util.Queue} for this task.
* @param inputQueue
*/
- public void addInputQueue(Queue<StreamsDatum> inputQueue);
+ public void addInputQueue(BlockingQueue<StreamsDatum> inputQueue);
/**
* Add an output {@link java.util.Queue} for this task.
* @param outputQueue
*/
- public void addOutputQueue(Queue<StreamsDatum> outputQueue);
+ public void addOutputQueue(BlockingQueue<StreamsDatum> outputQueue);
/**
* Set the configuration object that will shared and passed to all instances of StreamsTask.
@@ -73,12 +74,12 @@ public interface StreamsTask extends Runnable{
* Returns the input queues that have been set for this task.
* @return list of input queues
*/
- public List<Queue<StreamsDatum>> getInputQueues();
+ public List<BlockingQueue<StreamsDatum>> getInputQueues();
/**
* Returns the output queues that have been set for this task
* @return list of output queues
*/
- public List<Queue<StreamsDatum>> getOutputQueues();
+ public List<BlockingQueue<StreamsDatum>> getOutputQueues();
}
[05/11] git commit: Implemented and improved tests,
made fixes were tests exposed bugs
Posted by re...@apache.org.
Implemented and improved tests, made fixes were tests exposed bugs
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/0d207736
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/0d207736
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/0d207736
Branch: refs/heads/master
Commit: 0d20773648d12df3d99872c4fd1d377eca976e14
Parents: 1b18eb6
Author: Ryan Ebanks <ry...@gmail.com>
Authored: Wed Oct 8 12:50:00 2014 -0500
Committer: Ryan Ebanks <ry...@gmail.com>
Committed: Wed Oct 8 12:50:00 2014 -0500
----------------------------------------------------------------------
.../local/builders/LocalStreamBuilder.java | 16 ++-
.../streams/local/queues/ThroughputQueue.java | 17 +++-
.../streams/local/tasks/BaseStreamsTask.java | 12 +++
.../streams/local/tasks/StreamsMergeTask.java | 1 -
.../local/tasks/StreamsPersistWriterTask.java | 14 ++-
.../local/tasks/StreamsProcessorTask.java | 15 ++-
.../local/tasks/StreamsProviderTask.java | 5 +
.../apache/streams/local/tasks/StreamsTask.java | 5 +
.../local/builders/LocalStreamBuilderTest.java | 101 +++++++++++--------
.../PassthroughDatumCounterProcessor.java | 2 +
.../test/providers/NumericMessageProvider.java | 2 +-
11 files changed, 137 insertions(+), 53 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/0d207736/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 0b02f4e..4606aa7 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
@@ -33,6 +33,7 @@ import java.math.BigInteger;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
/**
@@ -48,6 +49,7 @@ public class LocalStreamBuilder implements StreamBuilder {
private Map<String, StreamComponent> providers;
private Map<String, StreamComponent> components;
private Map<String, Object> streamConfig;
+ private Map<StreamsTask, Future> futures;
private ExecutorService executor;
private ExecutorService monitor;
private int totalTasks;
@@ -100,6 +102,7 @@ public class LocalStreamBuilder implements StreamBuilder {
self.stopInternal(true);
}
};
+ this.futures = new HashMap<>();
}
@Override
@@ -177,6 +180,7 @@ public class LocalStreamBuilder implements StreamBuilder {
this.monitor = Executors.newFixedThreadPool(this.monitorTasks+1);
Map<String, StreamsProviderTask> provTasks = new HashMap<String, StreamsProviderTask>();
tasks = new HashMap<String, List<StreamsTask>>();
+ boolean forcedShutDown = false;
try {
monitorThread = new LocalStreamProcessMonitorThread(executor, 10);
this.monitor.submit(monitorThread);
@@ -198,8 +202,9 @@ public class LocalStreamBuilder implements StreamBuilder {
LOGGER.debug("Components are no longer running or timed out");
} catch (InterruptedException e){
LOGGER.warn("Runtime interrupted. Beginning shutdown");
+ forcedShutDown = true;
} finally{
- stop();
+ stopInternal(forcedShutDown);
}
}
@@ -216,10 +221,12 @@ public class LocalStreamBuilder implements StreamBuilder {
protected void forceShutdown(Map<String, List<StreamsTask>> streamsTasks) {
LOGGER.debug("Shutdown failed. Forcing shutdown");
- //give the stream 30secs to try to shutdown gracefully, then force shutdown otherwise
for(List<StreamsTask> tasks : streamsTasks.values()) {
for(StreamsTask task : tasks) {
task.stopTask();
+ if(task.isWaiting()) {
+ this.futures.get(task).cancel(true);
+ }
}
}
this.executor.shutdown();
@@ -277,7 +284,7 @@ public class LocalStreamBuilder implements StreamBuilder {
for(int i=0; i < tasks; ++i) {
StreamsTask task = comp.createConnectedTask(getTimeout());
task.setStreamConfig(this.streamConfig);
- this.executor.submit(task);
+ this.futures.put(task, this.executor.submit(task));
compTasks.add(task);
if( comp.isOperationCountable() ) {
this.monitor.submit(new StatusCounterMonitorThread((DatumStatusCountable) comp.getOperation(), 10));
@@ -311,6 +318,9 @@ public class LocalStreamBuilder implements StreamBuilder {
if(parentsShutDown) {
for(StreamsTask task : tasks) {
task.stopTask();
+ if(task.isWaiting()) {
+ this.futures.get(task).cancel(true); // no data to process, interrupt block queue
+ }
}
for(StreamsTask task : tasks) {
int count = 0;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/0d207736/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java
index 2bd27a8..579914b 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java
@@ -165,7 +165,22 @@ public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBe
@Override
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
- throw new NotImplementedException();
+ ThroughputElement<E> e = this.underlyingQueue.poll(timeout, unit);
+ if(e != null) {
+ try {
+ this.takeCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsRemoved;
+ Long queueTime = e.getWaited();
+ this.totalQueueTime += queueTime;
+ if(this.maxQueuedTime < queueTime) {
+ this.maxQueuedTime = queueTime;
+ }
+ } finally {
+ this.takeCountsLock.writeLock().unlock();
+ }
+ return e.getElement();
+ }
+ return null;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/0d207736/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java
index 758a883..902a2d7 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java
@@ -132,6 +132,18 @@ public abstract class BaseStreamsTask implements StreamsTask {
}
}
+ @Override
+ public boolean isWaiting() {
+ if(this.inQueues == null || this.inQueues.size() == 0) {
+ return true;
+ }
+ boolean empty = true;
+ for(Queue queue : this.inQueues) {
+ empty = empty && queue.isEmpty();
+ }
+ return empty;
+ }
+
/**
* //TODO LOCAL MODE HACK. Need to fix
* In order for our data streams to ported to other data flow frame works(Storm, Hadoop, Spark, etc) we need to be able to
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/0d207736/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsMergeTask.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsMergeTask.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsMergeTask.java
index 585e4dd..7a4c806 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsMergeTask.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsMergeTask.java
@@ -43,7 +43,6 @@ public class StreamsMergeTask extends BaseStreamsTask {
this.keepRunning = new AtomicBoolean(true);
}
-
@Override
public void stopTask() {
this.keepRunning.set(false);
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/0d207736/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 bb4f78f..42b32e4 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
@@ -28,6 +28,7 @@ import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
@@ -43,6 +44,7 @@ public class StreamsPersistWriterTask extends BaseStreamsTask implements DatumSt
private Map<String, Object> streamConfig;
private BlockingQueue<StreamsDatum> inQueue;
private AtomicBoolean isRunning;
+ private AtomicBoolean blocked;
private DatumStatusCounter statusCounter = new DatumStatusCounter();
@@ -70,6 +72,12 @@ public class StreamsPersistWriterTask extends BaseStreamsTask implements DatumSt
this.sleepTime = sleepTime;
this.keepRunning = new AtomicBoolean(true);
this.isRunning = new AtomicBoolean(true);
+ this.blocked = new AtomicBoolean(false);
+ }
+
+ @Override
+ public boolean isWaiting() {
+ return this.inQueue.isEmpty() && this.blocked.get();
}
@Override
@@ -94,7 +102,9 @@ public class StreamsPersistWriterTask extends BaseStreamsTask implements DatumSt
while(this.keepRunning.get()) {
StreamsDatum datum = null;
try {
- datum = this.inQueue.take();
+ this.blocked.set(true);
+ datum = this.inQueue.poll(5, TimeUnit.SECONDS);
+ this.blocked.set(false);
} catch (InterruptedException ie) {
LOGGER.error("Received InterruptedException. Shutting down and re-applying interrupt status.");
this.keepRunning.set(false);
@@ -111,7 +121,7 @@ public class StreamsPersistWriterTask extends BaseStreamsTask implements DatumSt
DatumUtils.addErrorToMetadata(datum, e, this.writer.getClass());
}
} else { //datums should never be null
- LOGGER.warn("Received null StreamsDatum @ writer : {}", this.writer.getClass().getName());
+ LOGGER.debug("Received null StreamsDatum @ writer : {}", this.writer.getClass().getName());
}
}
// StreamsDatum datum = this.inQueue.poll();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/0d207736/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 79cbf89..a3d367f 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
@@ -29,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
@@ -45,6 +46,7 @@ public class StreamsProcessorTask extends BaseStreamsTask implements DatumStatus
private Map<String, Object> streamConfig;
private BlockingQueue<StreamsDatum> inQueue;
private AtomicBoolean isRunning;
+ private AtomicBoolean blocked;
private DatumStatusCounter statusCounter = new DatumStatusCounter();
@@ -71,6 +73,12 @@ public class StreamsProcessorTask extends BaseStreamsTask implements DatumStatus
this.sleepTime = sleepTime;
this.keepRunning = new AtomicBoolean(true);
this.isRunning = new AtomicBoolean(true);
+ this.blocked = new AtomicBoolean(true);
+ }
+
+ @Override
+ public boolean isWaiting() {
+ return this.inQueue.isEmpty() && this.blocked.get();
}
@Override
@@ -100,9 +108,12 @@ public class StreamsProcessorTask extends BaseStreamsTask implements DatumStatus
while(this.keepRunning.get()) {
StreamsDatum datum = null;
try {
- datum = this.inQueue.take();
+ this.blocked.set(true);
+ datum = this.inQueue.poll(5, TimeUnit.SECONDS);
+ this.blocked.set(false);
} catch (InterruptedException ie) {
LOGGER.warn("Received InteruptedException, shutting down and re-applying interrupt status.");
+ this.keepRunning.set(false);
Thread.currentThread().interrupt();
}
if(datum != null) {
@@ -125,7 +136,7 @@ public class StreamsProcessorTask extends BaseStreamsTask implements DatumStatus
DatumUtils.addErrorToMetadata(datum, t, this.processor.getClass());
}
} else {
- LOGGER.warn("Removed NULL datum from queue at processor : {}", this.processor.getClass().getName());
+ LOGGER.debug("Removed NULL datum from queue at processor : {}", this.processor.getClass().getName());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/0d207736/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 e3f89c2..c16f64d 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
@@ -117,6 +117,11 @@ public class StreamsProviderTask extends BaseStreamsTask implements DatumStatusC
}
@Override
+ public boolean isWaiting() {
+ return false; //providers don't have inbound queues
+ }
+
+ @Override
public void stopTask() {
LOGGER.debug("Stopping Provider Task for {}", this.provider.getClass().getSimpleName());
this.keepRunning.set(false);
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/0d207736/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsTask.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsTask.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsTask.java
index 45c25f3..7513631 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsTask.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsTask.java
@@ -47,6 +47,11 @@ public interface StreamsTask extends Runnable{
public void stopTask();
/**
+ * Returns true if the task is waiting on more data to process
+ * @return true, if waiting on more data to process
+ */
+ public boolean isWaiting();
+ /**
* Add an input {@link java.util.Queue} for this task.
* @param inputQueue
*/
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/0d207736/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java
index 0e20e43..e602181 100644
--- a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java
+++ b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java
@@ -51,6 +51,7 @@ import org.apache.streams.local.test.providers.NumericMessageProvider;
import org.apache.streams.local.test.writer.DatumCounterWriter;
import org.apache.streams.local.test.writer.SystemOutWriter;
import org.junit.Before;
+import org.junit.Ignore;
import org.junit.Test;
import com.google.common.collect.Maps;
@@ -83,6 +84,7 @@ public class LocalStreamBuilderTest extends RandomizedTest {
}
}
+
@Test
public void testStreamIdValidations() {
StreamBuilder builder = new LocalStreamBuilder();
@@ -111,8 +113,18 @@ public class LocalStreamBuilderTest extends RandomizedTest {
}
@Test
- @Repeat(iterations = 3)
public void testBasicLinearStream2() {
+ linearStreamNonParallel(1004, 1);
+ }
+
+ @Test
+ public void testBasicLinearStream3() {
+ linearStreamNonParallel(1, 10);
+ }
+
+ @Test
+ @Repeat(iterations = 3)
+ public void testBasicLinearStreamRandom() {
int numDatums = randomIntBetween(1, 100000);
int numProcessors = randomIntBetween(1, 10);
linearStreamNonParallel(numDatums, numProcessors);
@@ -157,31 +169,32 @@ public class LocalStreamBuilderTest extends RandomizedTest {
@Test
public void testParallelLinearStream1() {
String processorId = "proc";
- int numProcessors = randomIntBetween(1, 100);
+ int numProcessors = randomIntBetween(1, 10);
int numDatums = randomIntBetween(1, 300000);
try {
- StreamBuilder builder = new LocalStreamBuilder();
+ StreamBuilder builder = new LocalStreamBuilder(50);
builder.newPerpetualStream("numeric_provider", new NumericMessageProvider(numDatums));
- AtomicInteger[] processorCounters = new AtomicInteger[numProcessors];
String connectTo = null;
for(int i=0; i < numProcessors; ++i) {
- processorCounters[i] = new AtomicInteger(0);
if(i == 0) {
connectTo = "numeric_provider";
} else {
connectTo = processorId+(i-1);
}
int parallelHint = randomIntBetween(1,5);
- builder.addStreamsProcessor(processorId+i, createPassThroughProcessor(processorCounters[i]), parallelHint, connectTo);
+ builder.addStreamsProcessor(processorId+i, new PassthroughDatumCounterProcessor(processorId+i), parallelHint, connectTo);
}
- Set output = Collections.newSetFromMap(new ConcurrentHashMap());
-
- builder.addStreamsPersistWriter("writer", createSetCollectingWriter(output), 1, processorId+(numProcessors-1));
+ builder.addStreamsPersistWriter("writer", new DatumCounterWriter("writer"), 1, processorId+(numProcessors-1));
builder.start();
- // can't test processors since they are serialized and deserialized when parallelized
+
+ assertEquals(numDatums, DatumCounterWriter.RECEIVED.get("writer").size());
for(int i=0; i < numDatums; ++i) {
- assertTrue("Expected writer to have received : "+i, output.contains(i));
+ assertTrue("Expected Writer to receive datum : " + i, DatumCounterWriter.RECEIVED.get("writer").contains(i));
+ }
+ for(int i=0; i < numProcessors; ++i) {
+ assertEquals(numDatums, PassthroughDatumCounterProcessor.COUNTS.get(processorId+i).get());
}
+
} finally {
for(int i=0; i < numProcessors; ++i) {
removeRegisteredMBeans(processorId+i);
@@ -194,23 +207,19 @@ public class LocalStreamBuilderTest extends RandomizedTest {
public void testBasicMergeStream() {
try {
int numDatums1 = randomIntBetween(1, 300000);
- int numDatums2 = randomIntBetween(1, 300000);;
- AtomicInteger counter1 = new AtomicInteger(0);
- AtomicInteger counter2 = new AtomicInteger(0);
- StreamsProcessor processor1 = createPassThroughProcessor(counter1);
- StreamsProcessor processor2 = createPassThroughProcessor(counter2);
+ int numDatums2 = randomIntBetween(1, 300000);
+ StreamsProcessor processor1 = new PassthroughDatumCounterProcessor("proc1");
+ StreamsProcessor processor2 = new PassthroughDatumCounterProcessor("proc2");
StreamBuilder builder = new LocalStreamBuilder();
- Set output = Collections.newSetFromMap(new ConcurrentHashMap());
- AtomicInteger outPutcounter = new AtomicInteger(0);
- builder.newReadCurrentStream("sp1", new NumericMessageProvider(numDatums1))
- .newReadCurrentStream("sp2", new NumericMessageProvider(numDatums2))
+ builder.newPerpetualStream("sp1", new NumericMessageProvider(numDatums1))
+ .newPerpetualStream("sp2", new NumericMessageProvider(numDatums2))
.addStreamsProcessor("proc1", processor1, 1, "sp1")
.addStreamsProcessor("proc2", processor2, 1, "sp2")
- .addStreamsPersistWriter("writer1", createSetCollectingWriter(output, outPutcounter), 1, "proc1", "proc2");
+ .addStreamsPersistWriter("writer1", new DatumCounterWriter("writer"), 1, "proc1", "proc2");
builder.start();
- assertEquals(numDatums1, counter1.get());
- assertEquals(numDatums2, counter2.get());
- assertEquals(numDatums1+numDatums2, outPutcounter.get());
+ assertEquals(numDatums1, PassthroughDatumCounterProcessor.COUNTS.get("proc1").get());
+ assertEquals(numDatums2, PassthroughDatumCounterProcessor.COUNTS.get("proc2").get());
+ assertEquals(numDatums1+numDatums2, DatumCounterWriter.COUNTS.get("writer").get());
} finally {
removeRegisteredMBeans("proc1", "proc2", "writer1");
}
@@ -220,19 +229,15 @@ public class LocalStreamBuilderTest extends RandomizedTest {
public void testBasicBranch() {
try {
int numDatums = randomIntBetween(1, 300000);
- StreamBuilder builder = new LocalStreamBuilder();
- AtomicInteger counter1 = new AtomicInteger(0);
- AtomicInteger counter2 = new AtomicInteger(0);
- Set output = Collections.newSetFromMap(new ConcurrentHashMap());
- AtomicInteger outputCounter = new AtomicInteger(0);
- builder.newReadCurrentStream("prov1", new NumericMessageProvider(numDatums))
- .addStreamsProcessor("proc1", createPassThroughProcessor(counter1), 1, "prov1")
- .addStreamsProcessor("proc2", createPassThroughProcessor(counter2), 1, "prov1")
- .addStreamsPersistWriter("w1", createSetCollectingWriter(output, outputCounter), 1, "proc1", "proc2");
+ StreamBuilder builder = new LocalStreamBuilder(50);
+ builder.newPerpetualStream("prov1", new NumericMessageProvider(numDatums))
+ .addStreamsProcessor("proc1", new PassthroughDatumCounterProcessor("proc1"), 1, "prov1")
+ .addStreamsProcessor("proc2", new PassthroughDatumCounterProcessor("proc2"), 1, "prov1")
+ .addStreamsPersistWriter("w1", new DatumCounterWriter("writer"), 1, "proc1", "proc2");
builder.start();
- assertEquals(numDatums, counter1.get());
- assertEquals(numDatums, counter2.get());
- assertEquals(numDatums*2, outputCounter.get());
+ assertEquals(numDatums, PassthroughDatumCounterProcessor.COUNTS.get("proc1").get());
+ assertEquals(numDatums, PassthroughDatumCounterProcessor.COUNTS.get("proc2").get());
+ assertEquals(numDatums*2, DatumCounterWriter.COUNTS.get("writer").get());
} finally {
removeRegisteredMBeans("prov1", "proc1", "proc2", "w1");
}
@@ -246,13 +251,11 @@ public class LocalStreamBuilderTest extends RandomizedTest {
Map<String, Object> config = Maps.newHashMap();
config.put(LocalStreamBuilder.TIMEOUT_KEY, timeout);
StreamBuilder builder = new LocalStreamBuilder(config);
- AtomicInteger counter = new AtomicInteger(0);
- Set output = Collections.newSetFromMap(new ConcurrentHashMap());
- builder.newReadCurrentStream("prov1", new NumericMessageProvider(numDatums))
+ builder.newPerpetualStream("prov1", new NumericMessageProvider(numDatums))
.addStreamsProcessor("proc1", new SlowProcessor(), 1, "prov1")
- .addStreamsPersistWriter("w1", createSetCollectingWriter(output, counter), 1, "proc1");
+ .addStreamsPersistWriter("w1", new DatumCounterWriter("writer"), 1, "proc1");
builder.start();
- assertEquals(numDatums, counter.get());
+ assertEquals(numDatums, DatumCounterWriter.COUNTS.get("writer").get());
} finally {
removeRegisteredMBeans("prov1", "proc1", "w1");
}
@@ -279,12 +282,13 @@ public class LocalStreamBuilderTest extends RandomizedTest {
}
}
+ @Ignore
@Test
public void ensureShutdownWithBlockedQueue() throws InterruptedException {
try {
ExecutorService service = Executors.newSingleThreadExecutor();
int before = Thread.activeCount();
- final StreamBuilder builder = new LocalStreamBuilder(1);
+ final StreamBuilder builder = new LocalStreamBuilder();
builder.newPerpetualStream("prov1", new NumericMessageProvider(30))
.addStreamsProcessor("proc1", new SlowProcessor(), 1, "prov1")
.addStreamsPersistWriter("w1", new SystemOutWriter(), 1, "proc1");
@@ -298,13 +302,24 @@ public class LocalStreamBuilderTest extends RandomizedTest {
Thread.sleep(500);
builder.stop();
service.shutdownNow();
- service.awaitTermination(1000, TimeUnit.MILLISECONDS);
+ service.awaitTermination(30000, TimeUnit.MILLISECONDS);
assertThat(Thread.activeCount(), is(equalTo(before)));
} finally {
removeRegisteredMBeans("prov1", "proc1", "w1");
}
}
+ @Before
+ private void clearCounters() {
+ PassthroughDatumCounterProcessor.COUNTS.clear();
+ PassthroughDatumCounterProcessor.CLAIMED_ID.clear();
+ PassthroughDatumCounterProcessor.SEEN_DATA.clear();
+ DatumCounterWriter.COUNTS.clear();
+ DatumCounterWriter.CLAIMED_ID.clear();
+ DatumCounterWriter.SEEN_DATA.clear();
+ DatumCounterWriter.RECEIVED.clear();
+ }
+
/**
* Creates {@link org.apache.streams.core.StreamsProcessor} that passes any StreamsDatum it gets as an
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/0d207736/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/processors/PassthroughDatumCounterProcessor.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/processors/PassthroughDatumCounterProcessor.java b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/processors/PassthroughDatumCounterProcessor.java
index b8249d1..bf1c338 100644
--- a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/processors/PassthroughDatumCounterProcessor.java
+++ b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/processors/PassthroughDatumCounterProcessor.java
@@ -78,6 +78,7 @@ public class PassthroughDatumCounterProcessor implements StreamsProcessor {
@Override
public void cleanUp() {
+ System.out.println("Clean up "+this.procId);
synchronized (COUNTS) {
AtomicLong count = COUNTS.get(this.procId);
if(count == null) {
@@ -86,6 +87,7 @@ public class PassthroughDatumCounterProcessor implements StreamsProcessor {
count.addAndGet(this.count);
}
}
+ System.out.println(this.procId+"\t"+this.count);
}
public int getMessageCount() {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/0d207736/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/providers/NumericMessageProvider.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/providers/NumericMessageProvider.java b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/providers/NumericMessageProvider.java
index 8a5de2d..ac8b6ff 100644
--- a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/providers/NumericMessageProvider.java
+++ b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/providers/NumericMessageProvider.java
@@ -63,7 +63,7 @@ public class NumericMessageProvider implements StreamsProvider {
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
}
- System.out.println("******************\n**\tBatchSize="+batch.size()+"\n******************");
+// System.out.println("******************\n**\tBatchSize="+batch.size()+"\n******************");
this.complete = batch.isEmpty() && this.data.isEmpty();
return new StreamsResultSet(batch);
}
[11/11] git commit: Merge branch 'STREAMS-191' of
github.com:rbnks/incubator-streams
Posted by re...@apache.org.
Merge branch 'STREAMS-191' of github.com:rbnks/incubator-streams
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/7b01eb48
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/7b01eb48
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/7b01eb48
Branch: refs/heads/master
Commit: 7b01eb48bdcafd61145e1801026aa45213e484ce
Parents: 0a32159 7882f14
Author: Ryan Ebanks <ry...@gmail.com>
Authored: Mon Oct 20 12:12:23 2014 -0500
Committer: Ryan Ebanks <ry...@gmail.com>
Committed: Mon Oct 20 12:12:23 2014 -0500
----------------------------------------------------------------------
.../local/builders/LocalStreamBuilder.java | 58 +--
.../streams/local/builders/StreamComponent.java | 21 +-
.../streams/local/tasks/BaseStreamsTask.java | 52 ++-
.../streams/local/tasks/StreamsMergeTask.java | 9 +-
.../local/tasks/StreamsPersistWriterTask.java | 43 +-
.../local/tasks/StreamsProcessorTask.java | 52 ++-
.../local/tasks/StreamsProviderTask.java | 8 +-
.../apache/streams/local/tasks/StreamsTask.java | 14 +-
.../local/builders/LocalStreamBuilderTest.java | 399 ++++++++++++-------
.../local/builders/ToyLocalBuilderExample.java | 5 +-
.../streams/local/tasks/BasicTasksTest.java | 52 +--
.../local/tasks/StreamsProviderTaskTest.java | 7 +-
.../PassthroughDatumCounterProcessor.java | 51 ++-
.../test/providers/NumericMessageProvider.java | 29 +-
.../local/test/writer/DatumCounterWriter.java | 61 ++-
15 files changed, 569 insertions(+), 292 deletions(-)
----------------------------------------------------------------------
[08/11] git commit: Merge pull request #6 from apache/master
Posted by re...@apache.org.
Merge pull request #6 from apache/master
Merge Apache Master 2014/10/13
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/4704012f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/4704012f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/4704012f
Branch: refs/heads/master
Commit: 4704012fd985cffbf60c37aa2e57f8f829ff6f9e
Parents: c483941 6dd1ea5
Author: Ryan Ebanks <ry...@raveldata.com>
Authored: Mon Oct 13 10:57:21 2014 -0500
Committer: Ryan Ebanks <ry...@raveldata.com>
Committed: Mon Oct 13 10:57:21 2014 -0500
----------------------------------------------------------------------
pom.xml | 9 +-
streams-config/pom.xml | 1 -
.../streams-persist-s3/pom.xml | 1 -
streams-contrib/streams-persist-hdfs/pom.xml | 13 +-
.../apache/streams/hdfs/HdfsConfigurator.java | 38 +-
.../streams/hdfs/WebHdfsPersistReader.java | 13 +-
.../streams/hdfs/WebHdfsPersistReaderTask.java | 5 +-
.../streams/hdfs/WebHdfsPersistWriter.java | 14 +-
.../apache/streams/hdfs/HdfsConfiguration.json | 6 +
.../streams/hdfs/HdfsWriterConfiguration.json | 5 +
.../hdfs/test/HdfsPersistConfigTest.java | 169 +++
streams-contrib/streams-persist-kafka/pom.xml | 1 -
streams-contrib/streams-persist-mongo/pom.xml | 1 -
.../streams-provider-datasift/pom.xml | 2 -
.../streams-provider-facebook/pom.xml | 1 -
.../gnip-edc-facebook/pom.xml | 1 -
.../gnip-edc-youtube/pom.xml | 3 +-
.../gnip-powertrack/pom.xml | 39 +-
.../google-gmail/pom.xml | 1 -
.../google-gplus/pom.xml | 1 -
.../streams-provider-instagram/pom.xml | 1 -
.../streams-provider-moreover/pom.xml | 3 +-
streams-contrib/streams-provider-rss/pom.xml | 50 +-
.../streams/rss/provider/RssStreamProvider.java | 169 ++-
.../rss/provider/RssStreamProviderTask.java | 214 +++-
.../provider/perpetual/RssFeedScheduler.java | 112 ++
.../rss/provider/RssStreamProviderTaskTest.java | 148 +++
.../rss/provider/RssStreamProviderTest.java | 103 ++
.../perpetual/RssFeedSchedulerTest.java | 101 ++
.../test/resources/test_rss_xml/economist1.xml | 233 ++++
.../test/resources/test_rss_xml/economist2.xml | 1069 ++++++++++++++++++
.../test_rss_xml/economistCombined.xml | 221 ++++
.../streams-provider-sysomos/pom.xml | 1 -
.../streams-provider-twitter/pom.xml | 1 -
.../provider/TwitterTimelineProvider.java | 10 +-
.../provider/TwitterTimelineProviderTask.java | 6 +-
streams-pojo/pom.xml | 1 -
.../streams/local/queues/ThroughputQueue.java | 440 +++++++
.../local/queues/ThroughputQueueMXBean.java | 66 ++
.../local/tasks/StreamsProviderTask.java | 11 +-
.../queues/ThroughputQueueMulitThreadTest.java | 302 +++++
.../queues/ThroughputQueueSingleThreadTest.java | 243 ++++
42 files changed, 3597 insertions(+), 232 deletions(-)
----------------------------------------------------------------------
[03/11] git commit: Partial Commit. Tests are taking too long due to
poor provider design
Posted by re...@apache.org.
Partial Commit. Tests are taking too long due to poor provider design
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/c2ade3ad
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/c2ade3ad
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/c2ade3ad
Branch: refs/heads/master
Commit: c2ade3adfbd9c09c33de8941e81f2cd8fc849db2
Parents: 3fad21a
Author: Ryan Ebanks <ry...@gmail.com>
Authored: Tue Oct 7 15:40:49 2014 -0500
Committer: Ryan Ebanks <ry...@gmail.com>
Committed: Tue Oct 7 15:40:49 2014 -0500
----------------------------------------------------------------------
.../local/tasks/StreamsPersistWriterTask.java | 1 +
.../local/tasks/StreamsProcessorTask.java | 1 +
.../local/builders/LocalStreamBuilderTest.java | 384 ++++++++++++-------
.../streams/local/tasks/BasicTasksTest.java | 8 +-
.../PassthroughDatumCounterProcessor.java | 49 ++-
.../test/providers/NumericMessageProvider.java | 29 +-
.../local/test/writer/DatumCounterWriter.java | 61 ++-
7 files changed, 365 insertions(+), 168 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/c2ade3ad/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 ce9f5a7..bb4f78f 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
@@ -97,6 +97,7 @@ public class StreamsPersistWriterTask extends BaseStreamsTask implements DatumSt
datum = this.inQueue.take();
} catch (InterruptedException ie) {
LOGGER.error("Received InterruptedException. Shutting down and re-applying interrupt status.");
+ this.keepRunning.set(false);
Thread.currentThread().interrupt();
}
if(datum != null) {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/c2ade3ad/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 4b65066..79cbf89 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
@@ -116,6 +116,7 @@ public class StreamsProcessorTask extends BaseStreamsTask implements DatumStatus
}
} catch (InterruptedException ie) {
LOGGER.warn("Received InteruptedException, shutting down and re-applying interrupt status.");
+ this.keepRunning.set(false);
Thread.currentThread().interrupt();
} catch (Throwable t) {
LOGGER.warn("Caught Throwable in processor, {} : {}", this.processor.getClass().getName(), t.getMessage());
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/c2ade3ad/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java
index b082168..0e20e43 100644
--- a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java
+++ b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java
@@ -18,39 +18,47 @@
package org.apache.streams.local.builders;
-import static org.hamcrest.Matchers.allOf;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.greaterThan;
-import static org.hamcrest.Matchers.greaterThanOrEqualTo;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.lessThanOrEqualTo;
+import static org.hamcrest.Matchers.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Scanner;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
+import java.lang.management.ManagementFactory;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import com.carrotsearch.randomizedtesting.RandomizedTest;
+import com.carrotsearch.randomizedtesting.annotations.Repeat;
+import com.google.common.collect.Lists;
import org.apache.streams.core.StreamBuilder;
import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsPersistWriter;
+import org.apache.streams.core.StreamsProcessor;
+import org.apache.streams.local.queues.ThroughputQueue;
import org.apache.streams.local.test.processors.PassthroughDatumCounterProcessor;
import org.apache.streams.local.test.processors.SlowProcessor;
import org.apache.streams.local.test.providers.EmptyResultSetProvider;
import org.apache.streams.local.test.providers.NumericMessageProvider;
+import org.apache.streams.local.test.writer.DatumCounterWriter;
import org.apache.streams.local.test.writer.SystemOutWriter;
import org.junit.Before;
import org.junit.Test;
import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import javax.management.*;
/**
* Basic Tests for the LocalStreamBuilder.
@@ -61,14 +69,18 @@ import com.google.common.collect.Queues;
*
*
*/
-public class LocalStreamBuilderTest {
+public class LocalStreamBuilderTest extends RandomizedTest {
- ByteArrayOutputStream out;
- @Before
- public void setSystemOut() {
- out = new ByteArrayOutputStream();
- System.setOut(new PrintStream(out));
+ public void removeRegisteredMBeans(String... ids) {
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ for(String id : ids) {
+ try {
+ mbs.unregisterMBean(new ObjectName(String.format(ThroughputQueue.NAME_TEMPLATE, id)));
+ } catch (MalformedObjectNameException|InstanceNotFoundException|MBeanRegistrationException e) {
+ //No-op
+ }
+ }
}
@Test
@@ -83,176 +95,260 @@ public class LocalStreamBuilderTest {
}
assertNotNull(exp);
exp = null;
- builder.addStreamsProcessor("1", new PassthroughDatumCounterProcessor(), 1, "id");
+ builder.addStreamsProcessor("1", new PassthroughDatumCounterProcessor("1"), 1, "id");
try {
- builder.addStreamsProcessor("2", new PassthroughDatumCounterProcessor(), 1, "id", "id2");
+ builder.addStreamsProcessor("2", new PassthroughDatumCounterProcessor("2"), 1, "id", "id2");
} catch (RuntimeException e) {
exp = e;
}
assertNotNull(exp);
+ removeRegisteredMBeans("1", "2");
}
@Test
public void testBasicLinearStream1() {
- int numDatums = 1;
- StreamBuilder builder = new LocalStreamBuilder();
- PassthroughDatumCounterProcessor processor = new PassthroughDatumCounterProcessor();
- SystemOutWriter writer = new SystemOutWriter();
- builder.newReadCurrentStream("sp1", new NumericMessageProvider(numDatums))
- .addStreamsProcessor("proc1", processor, 1, "sp1")
- .addStreamsPersistWriter("writer1", writer, 1, "proc1");
- builder.start();
- int count = 0;
- Scanner scanner = new Scanner(new ByteArrayInputStream(out.toByteArray()));
- while(scanner.hasNextLine()) {
- ++count;
- scanner.nextLine();
- }
- assertThat(count, greaterThan(numDatums)); // using > because number of lines in system.out is non-deterministic
+ linearStreamNonParallel(1, 1);
}
@Test
+ @Repeat(iterations = 3)
public void testBasicLinearStream2() {
- int numDatums = 100;
- StreamBuilder builder = new LocalStreamBuilder();
- PassthroughDatumCounterProcessor processor = new PassthroughDatumCounterProcessor();
- SystemOutWriter writer = new SystemOutWriter();
- builder.newReadCurrentStream("sp1", new NumericMessageProvider(numDatums))
- .addStreamsProcessor("proc1", processor, 1, "sp1")
- .addStreamsPersistWriter("writer1", writer, 1, "proc1");
- builder.start();
- int count = 0;
- Scanner scanner = new Scanner(new ByteArrayInputStream(out.toByteArray()));
- while(scanner.hasNextLine()) {
- ++count;
- scanner.nextLine();
+ int numDatums = randomIntBetween(1, 100000);
+ int numProcessors = randomIntBetween(1, 10);
+ linearStreamNonParallel(numDatums, numProcessors);
+ }
+
+ /**
+ * Tests that all datums pass through each processor and that all datums reach the writer
+ * @param numDatums
+ * @param numProcessors
+ */
+ private void linearStreamNonParallel(int numDatums, int numProcessors) {
+ String processorId = "proc";
+ try {
+ StreamBuilder builder = new LocalStreamBuilder(10);
+ builder.newPerpetualStream("numeric_provider", new NumericMessageProvider(numDatums));
+ String connectTo = null;
+ for(int i=0; i < numProcessors; ++i) {
+ if(i == 0) {
+ connectTo = "numeric_provider";
+ } else {
+ connectTo = processorId+(i-1);
+ }
+ builder.addStreamsProcessor(processorId+i, new PassthroughDatumCounterProcessor(processorId+i), 1, connectTo);
+ }
+ Set output = Collections.newSetFromMap(new ConcurrentHashMap());
+ builder.addStreamsPersistWriter("writer", new DatumCounterWriter("writer"), 1, processorId+(numProcessors-1));
+ builder.start();
+ for(int i=0; i < numProcessors; ++i) {
+ assertEquals("Processor "+i+" did not receive all of the datums", numDatums, PassthroughDatumCounterProcessor.COUNTS.get(processorId+i).get());
+ }
+ for(int i=0; i < numDatums; ++i) {
+ assertTrue("Expected writer to have received : "+i, DatumCounterWriter.RECEIVED.get("writer").contains(i));
+ }
+ } finally {
+ for(int i=0; i < numProcessors; ++i) {
+ removeRegisteredMBeans(processorId+i);
+ }
+ removeRegisteredMBeans("writer");
}
- assertThat(count, greaterThan(numDatums)); // using > because number of lines in system.out is non-deterministic
}
@Test
public void testParallelLinearStream1() {
- int numDatums = 1000;
- int parallelHint = 20;
- PassthroughDatumCounterProcessor.sawData = new HashSet<Integer>();
- PassthroughDatumCounterProcessor.claimedNumber = new HashSet<Integer>();
- StreamBuilder builder = new LocalStreamBuilder();
- PassthroughDatumCounterProcessor processor = new PassthroughDatumCounterProcessor();
- SystemOutWriter writer = new SystemOutWriter();
- builder.newReadCurrentStream("sp1", new NumericMessageProvider(numDatums))
- .addStreamsProcessor("proc1", processor, parallelHint, "sp1")
- .addStreamsPersistWriter("writer1", writer, 1, "proc1");
- builder.start();
- int count = 0;
- Scanner scanner = new Scanner(new ByteArrayInputStream(out.toByteArray()));
- while(scanner.hasNextLine()) {
- ++count;
- scanner.nextLine();
+ String processorId = "proc";
+ int numProcessors = randomIntBetween(1, 100);
+ int numDatums = randomIntBetween(1, 300000);
+ try {
+ StreamBuilder builder = new LocalStreamBuilder();
+ builder.newPerpetualStream("numeric_provider", new NumericMessageProvider(numDatums));
+ AtomicInteger[] processorCounters = new AtomicInteger[numProcessors];
+ String connectTo = null;
+ for(int i=0; i < numProcessors; ++i) {
+ processorCounters[i] = new AtomicInteger(0);
+ if(i == 0) {
+ connectTo = "numeric_provider";
+ } else {
+ connectTo = processorId+(i-1);
+ }
+ int parallelHint = randomIntBetween(1,5);
+ builder.addStreamsProcessor(processorId+i, createPassThroughProcessor(processorCounters[i]), parallelHint, connectTo);
+ }
+ Set output = Collections.newSetFromMap(new ConcurrentHashMap());
+
+ builder.addStreamsPersistWriter("writer", createSetCollectingWriter(output), 1, processorId+(numProcessors-1));
+ builder.start();
+ // can't test processors since they are serialized and deserialized when parallelized
+ for(int i=0; i < numDatums; ++i) {
+ assertTrue("Expected writer to have received : "+i, output.contains(i));
+ }
+ } finally {
+ for(int i=0; i < numProcessors; ++i) {
+ removeRegisteredMBeans(processorId+i);
+ }
+ removeRegisteredMBeans("writer");
}
- assertThat(count, greaterThan(numDatums)); // using > because number of lines in system.out is non-deterministic
- assertEquals(parallelHint, PassthroughDatumCounterProcessor.claimedNumber.size()); //test 40 were initialized
- assertTrue(PassthroughDatumCounterProcessor.sawData.size() > 1 && PassthroughDatumCounterProcessor.sawData.size() <= parallelHint); //test more than one processor got data
}
@Test
public void testBasicMergeStream() {
- int numDatums1 = 1;
- int numDatums2 = 100;
- PassthroughDatumCounterProcessor processor1 = new PassthroughDatumCounterProcessor();
- PassthroughDatumCounterProcessor processor2 = new PassthroughDatumCounterProcessor();
- SystemOutWriter writer = new SystemOutWriter();
- StreamBuilder builder = new LocalStreamBuilder();
- builder.newReadCurrentStream("sp1", new NumericMessageProvider(numDatums1))
- .newReadCurrentStream("sp2", new NumericMessageProvider(numDatums2))
- .addStreamsProcessor("proc1", processor1, 1, "sp1")
- .addStreamsProcessor("proc2", processor2, 1, "sp2")
- .addStreamsPersistWriter("writer1", writer, 1, "proc1", "proc2");
- builder.start();
- int count = 0;
- Scanner scanner = new Scanner(new ByteArrayInputStream(out.toByteArray()));
- while (scanner.hasNextLine()) {
- ++count;
- scanner.nextLine();
+ try {
+ int numDatums1 = randomIntBetween(1, 300000);
+ int numDatums2 = randomIntBetween(1, 300000);;
+ AtomicInteger counter1 = new AtomicInteger(0);
+ AtomicInteger counter2 = new AtomicInteger(0);
+ StreamsProcessor processor1 = createPassThroughProcessor(counter1);
+ StreamsProcessor processor2 = createPassThroughProcessor(counter2);
+ StreamBuilder builder = new LocalStreamBuilder();
+ Set output = Collections.newSetFromMap(new ConcurrentHashMap());
+ AtomicInteger outPutcounter = new AtomicInteger(0);
+ builder.newReadCurrentStream("sp1", new NumericMessageProvider(numDatums1))
+ .newReadCurrentStream("sp2", new NumericMessageProvider(numDatums2))
+ .addStreamsProcessor("proc1", processor1, 1, "sp1")
+ .addStreamsProcessor("proc2", processor2, 1, "sp2")
+ .addStreamsPersistWriter("writer1", createSetCollectingWriter(output, outPutcounter), 1, "proc1", "proc2");
+ builder.start();
+ assertEquals(numDatums1, counter1.get());
+ assertEquals(numDatums2, counter2.get());
+ assertEquals(numDatums1+numDatums2, outPutcounter.get());
+ } finally {
+ removeRegisteredMBeans("proc1", "proc2", "writer1");
}
- assertThat(count, greaterThan(numDatums1 + numDatums2)); // using > because number of lines in system.out is non-deterministic
}
@Test
public void testBasicBranch() {
- int numDatums = 100;
- StreamBuilder builder = new LocalStreamBuilder();
- builder.newReadCurrentStream("prov1", new NumericMessageProvider(numDatums))
- .addStreamsProcessor("proc1", new PassthroughDatumCounterProcessor(), 1, "prov1")
- .addStreamsProcessor("proc2", new PassthroughDatumCounterProcessor(), 1, "prov1")
- .addStreamsPersistWriter("w1", new SystemOutWriter(), 1, "proc1", "proc2");
- builder.start();
- int count = 0;
- Scanner scanner = new Scanner(new ByteArrayInputStream(out.toByteArray()));
- while(scanner.hasNextLine()) {
- ++count;
- scanner.nextLine();
+ try {
+ int numDatums = randomIntBetween(1, 300000);
+ StreamBuilder builder = new LocalStreamBuilder();
+ AtomicInteger counter1 = new AtomicInteger(0);
+ AtomicInteger counter2 = new AtomicInteger(0);
+ Set output = Collections.newSetFromMap(new ConcurrentHashMap());
+ AtomicInteger outputCounter = new AtomicInteger(0);
+ builder.newReadCurrentStream("prov1", new NumericMessageProvider(numDatums))
+ .addStreamsProcessor("proc1", createPassThroughProcessor(counter1), 1, "prov1")
+ .addStreamsProcessor("proc2", createPassThroughProcessor(counter2), 1, "prov1")
+ .addStreamsPersistWriter("w1", createSetCollectingWriter(output, outputCounter), 1, "proc1", "proc2");
+ builder.start();
+ assertEquals(numDatums, counter1.get());
+ assertEquals(numDatums, counter2.get());
+ assertEquals(numDatums*2, outputCounter.get());
+ } finally {
+ removeRegisteredMBeans("prov1", "proc1", "proc2", "w1");
}
- assertThat(count, greaterThan(numDatums * 2)); // using > because number of lines in system.out is non-deterministic
-
}
@Test
public void testSlowProcessorBranch() {
- int numDatums = 30;
- int timeout = 2000;
- Map<String, Object> config = Maps.newHashMap();
- config.put(LocalStreamBuilder.TIMEOUT_KEY, timeout);
- StreamBuilder builder = new LocalStreamBuilder(config);
- builder.newReadCurrentStream("prov1", new NumericMessageProvider(numDatums))
- .addStreamsProcessor("proc1", new SlowProcessor(), 1, "prov1")
- .addStreamsPersistWriter("w1", new SystemOutWriter(), 1, "proc1");
- builder.start();
- int count = 0;
- Scanner scanner = new Scanner(new ByteArrayInputStream(out.toByteArray()));
- while(scanner.hasNextLine()) {
- ++count;
- scanner.nextLine();
+ try {
+ int numDatums = 30;
+ int timeout = 2000;
+ Map<String, Object> config = Maps.newHashMap();
+ config.put(LocalStreamBuilder.TIMEOUT_KEY, timeout);
+ StreamBuilder builder = new LocalStreamBuilder(config);
+ AtomicInteger counter = new AtomicInteger(0);
+ Set output = Collections.newSetFromMap(new ConcurrentHashMap());
+ builder.newReadCurrentStream("prov1", new NumericMessageProvider(numDatums))
+ .addStreamsProcessor("proc1", new SlowProcessor(), 1, "prov1")
+ .addStreamsPersistWriter("w1", createSetCollectingWriter(output, counter), 1, "proc1");
+ builder.start();
+ assertEquals(numDatums, counter.get());
+ } finally {
+ removeRegisteredMBeans("prov1", "proc1", "w1");
}
- assertThat(count, greaterThan(numDatums)); // using > because number of lines in system.out is non-deterministic
-
}
@Test
public void testConfiguredProviderTimeout() {
- Map<String, Object> config = Maps.newHashMap();
- int timeout = 10000;
- config.put(LocalStreamBuilder.TIMEOUT_KEY, timeout);
- long start = System.currentTimeMillis();
- StreamBuilder builder = new LocalStreamBuilder(-1, config);
- builder.newPerpetualStream("prov1", new EmptyResultSetProvider())
- .addStreamsProcessor("proc1", new PassthroughDatumCounterProcessor(), 1, "prov1")
- .addStreamsProcessor("proc2", new PassthroughDatumCounterProcessor(), 1, "proc1")
- .addStreamsPersistWriter("w1", new SystemOutWriter(), 1, "proc1");
- builder.start();
- long end = System.currentTimeMillis();
- //We care mostly that it doesn't terminate too early. With thread shutdowns, etc, the actual time is indeterminate. Just make sure there is an upper bound
- assertThat((int)(end - start), is(allOf(greaterThanOrEqualTo(timeout), lessThanOrEqualTo(4 * timeout))));
+ try {
+ Map<String, Object> config = Maps.newHashMap();
+ int timeout = 10000;
+ config.put(LocalStreamBuilder.TIMEOUT_KEY, timeout);
+ long start = System.currentTimeMillis();
+ StreamBuilder builder = new LocalStreamBuilder(-1, config);
+ builder.newPerpetualStream("prov1", new EmptyResultSetProvider())
+ .addStreamsProcessor("proc1", new PassthroughDatumCounterProcessor("proc1"), 1, "prov1")
+ .addStreamsProcessor("proc2", new PassthroughDatumCounterProcessor("proc2"), 1, "proc1")
+ .addStreamsPersistWriter("w1", new SystemOutWriter(), 1, "proc1");
+ builder.start();
+ long end = System.currentTimeMillis();
+ //We care mostly that it doesn't terminate too early. With thread shutdowns, etc, the actual time is indeterminate. Just make sure there is an upper bound
+ assertThat((int) (end - start), is(allOf(greaterThanOrEqualTo(timeout), lessThanOrEqualTo(4 * timeout))));
+ } finally {
+ removeRegisteredMBeans("prov1", "proc1", "proc2", "w1");
+ }
}
@Test
public void ensureShutdownWithBlockedQueue() throws InterruptedException {
- ExecutorService service = Executors.newSingleThreadExecutor();
- int before = Thread.activeCount();
- final StreamBuilder builder = new LocalStreamBuilder(1);
- builder.newPerpetualStream("prov1", new NumericMessageProvider(30))
- .addStreamsProcessor("proc1", new SlowProcessor(), 1, "prov1")
- .addStreamsPersistWriter("w1", new SystemOutWriter(), 1, "proc1");
- service.submit(new Runnable(){
+ try {
+ ExecutorService service = Executors.newSingleThreadExecutor();
+ int before = Thread.activeCount();
+ final StreamBuilder builder = new LocalStreamBuilder(1);
+ builder.newPerpetualStream("prov1", new NumericMessageProvider(30))
+ .addStreamsProcessor("proc1", new SlowProcessor(), 1, "prov1")
+ .addStreamsPersistWriter("w1", new SystemOutWriter(), 1, "proc1");
+ service.submit(new Runnable() {
+ @Override
+ public void run() {
+ builder.start();
+ }
+ });
+ //Let streams spin up threads and start to process
+ Thread.sleep(500);
+ builder.stop();
+ service.shutdownNow();
+ service.awaitTermination(1000, TimeUnit.MILLISECONDS);
+ assertThat(Thread.activeCount(), is(equalTo(before)));
+ } finally {
+ removeRegisteredMBeans("prov1", "proc1", "w1");
+ }
+ }
+
+
+ /**
+ * Creates {@link org.apache.streams.core.StreamsProcessor} that passes any StreamsDatum it gets as an
+ * input and counts the number of items it processes.
+ * @param counter
+ * @return
+ */
+ private StreamsProcessor createPassThroughProcessor(final AtomicInteger counter) {
+ StreamsProcessor processor = mock(StreamsProcessor.class);
+ when(processor.process(any(StreamsDatum.class))).thenAnswer(new Answer<List<StreamsDatum>>() {
@Override
- public void run() {
- builder.start();
+ public List<StreamsDatum> answer(InvocationOnMock invocationOnMock) throws Throwable {
+ List<StreamsDatum> datum = Lists.newLinkedList();
+ if(counter != null) {
+ counter.incrementAndGet();
+ }
+ datum.add((StreamsDatum) invocationOnMock.getArguments()[0] );
+ return datum;
}
});
- //Let streams spin up threads and start to process
- Thread.sleep(500);
- builder.stop();
- service.shutdownNow();
- service.awaitTermination(1000, TimeUnit.MILLISECONDS);
- assertThat(Thread.activeCount(), is(equalTo(before)));
+ return processor;
+ }
+
+ private StreamsPersistWriter createSetCollectingWriter(final Set collector) {
+ return createSetCollectingWriter(collector, null);
+ }
+
+ /**
+ * Creates a StreamsPersistWriter that adds every datums document to a set
+ * @param collector
+ * @return
+ */
+ private StreamsPersistWriter createSetCollectingWriter(final Set collector, final AtomicInteger counter) {
+ StreamsPersistWriter writer = mock(StreamsPersistWriter.class);
+ doAnswer(new Answer() {
+ @Override
+ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+ if(counter != null) {
+ counter.incrementAndGet();
+ }
+ collector.add(((StreamsDatum)invocationOnMock.getArguments()[0]).getDocument());
+ return null;
+ }
+ }).when(writer).write(any(StreamsDatum.class));
+ return writer;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/c2ade3ad/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/BasicTasksTest.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/BasicTasksTest.java b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/BasicTasksTest.java
index c4ea4ee..f524db0 100644
--- a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/BasicTasksTest.java
+++ b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/BasicTasksTest.java
@@ -83,7 +83,7 @@ public class BasicTasksTest {
@Test
public void testProcessorTask() {
int numMessages = 100;
- PassthroughDatumCounterProcessor processor = new PassthroughDatumCounterProcessor();
+ PassthroughDatumCounterProcessor processor = new PassthroughDatumCounterProcessor("");
StreamsProcessorTask task = new StreamsProcessorTask(processor);
BlockingQueue<StreamsDatum> outQueue = new LinkedBlockingQueue<>();
BlockingQueue<StreamsDatum> inQueue = createInputQueue(numMessages);
@@ -121,7 +121,7 @@ public class BasicTasksTest {
@Test
public void testWriterTask() {
int numMessages = 100;
- DatumCounterWriter writer = new DatumCounterWriter();
+ DatumCounterWriter writer = new DatumCounterWriter("");
StreamsPersistWriterTask task = new StreamsPersistWriterTask(writer);
BlockingQueue<StreamsDatum> outQueue = new LinkedBlockingQueue<>();
BlockingQueue<StreamsDatum> inQueue = createInputQueue(numMessages);
@@ -203,7 +203,7 @@ public class BasicTasksTest {
@Test
public void testBranching() {
int numMessages = 100;
- PassthroughDatumCounterProcessor processor = new PassthroughDatumCounterProcessor();
+ PassthroughDatumCounterProcessor processor = new PassthroughDatumCounterProcessor("");
StreamsProcessorTask task = new StreamsProcessorTask(processor);
BlockingQueue<StreamsDatum> outQueue1 = new LinkedBlockingQueue<>();
BlockingQueue<StreamsDatum> outQueue2 = new LinkedBlockingQueue<>();
@@ -246,7 +246,7 @@ public class BasicTasksTest {
@Test
public void testBranchingSerialization() {
int numMessages = 1;
- PassthroughDatumCounterProcessor processor = new PassthroughDatumCounterProcessor();
+ PassthroughDatumCounterProcessor processor = new PassthroughDatumCounterProcessor("");
StreamsProcessorTask task = new StreamsProcessorTask(processor);
BlockingQueue<StreamsDatum> outQueue1 = new LinkedBlockingQueue<>();
BlockingQueue<StreamsDatum> outQueue2 = new LinkedBlockingQueue<>();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/c2ade3ad/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/processors/PassthroughDatumCounterProcessor.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/processors/PassthroughDatumCounterProcessor.java b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/processors/PassthroughDatumCounterProcessor.java
index 423c05a..b8249d1 100644
--- a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/processors/PassthroughDatumCounterProcessor.java
+++ b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/processors/PassthroughDatumCounterProcessor.java
@@ -22,43 +22,70 @@ import org.apache.streams.core.StreamsDatum;
import org.apache.streams.core.StreamsProcessor;
import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
/**
- * Created by rebanks on 2/18/14.
+ *
*/
public class PassthroughDatumCounterProcessor implements StreamsProcessor {
- public static Set<Integer> claimedNumber = new HashSet<Integer>();
- public static final Random rand = new Random();
- public static Set<Integer> sawData = new HashSet<Integer>();
+ /**
+ * Set of all ids that have been claimed. Ensures all instances are assigned unique ids
+ */
+ public static Set<Integer> CLAIMED_ID = new HashSet<Integer>();
+ /**
+ * Random instance to generate ids
+ */
+ public static final Random RAND = new Random();
+ /**
+ * Set of instance ids that received data. Usefully for testing parrallelization is actually working.
+ */
+ public final static Set<Integer> SEEN_DATA = new HashSet<Integer>();
+ /**
+ * The total count of data seen by a all instances of a processor.
+ */
+ public static final ConcurrentHashMap<String, AtomicLong> COUNTS = new ConcurrentHashMap<>();
private int count = 0;
private int id;
+ private String procId;
+
+ public PassthroughDatumCounterProcessor(String procId) {
+ this.procId = procId;
+ }
@Override
public List<StreamsDatum> process(StreamsDatum entry) {
++this.count;
List<StreamsDatum> result = new LinkedList<StreamsDatum>();
result.add(entry);
- synchronized (sawData) {
- sawData.add(this.id);
+ synchronized (SEEN_DATA) {
+ SEEN_DATA.add(this.id);
}
return result;
}
@Override
public void prepare(Object configurationObject) {
- synchronized (claimedNumber) {
- this.id = rand.nextInt();
- while(!claimedNumber.add(this.id)) {
- this.id = rand.nextInt();
+ synchronized (CLAIMED_ID) {
+ this.id = RAND.nextInt();
+ while(!CLAIMED_ID.add(this.id)) {
+ this.id = RAND.nextInt();
}
}
}
@Override
public void cleanUp() {
-
+ synchronized (COUNTS) {
+ AtomicLong count = COUNTS.get(this.procId);
+ if(count == null) {
+ COUNTS.put(this.procId, new AtomicLong(this.count));
+ } else {
+ count.addAndGet(this.count);
+ }
+ }
}
public int getMessageCount() {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/c2ade3ad/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/providers/NumericMessageProvider.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/providers/NumericMessageProvider.java b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/providers/NumericMessageProvider.java
index f076e4b..8a5de2d 100644
--- a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/providers/NumericMessageProvider.java
+++ b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/providers/NumericMessageProvider.java
@@ -27,6 +27,7 @@ import org.joda.time.DateTime;
import java.math.BigInteger;
import java.util.Iterator;
import java.util.Queue;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -35,7 +36,11 @@ import java.util.concurrent.atomic.AtomicBoolean;
*/
public class NumericMessageProvider implements StreamsProvider {
+ private static final int DEFAULT_BATCH_SIZE = 100;
+
private int numMessages;
+ private BlockingQueue<StreamsDatum> data;
+ private volatile boolean complete = false;
public NumericMessageProvider(int numMessages) {
this.numMessages = numMessages;
@@ -43,12 +48,24 @@ public class NumericMessageProvider implements StreamsProvider {
@Override
public void startStream() {
- // no op
+ this.data = constructQueue();
}
@Override
public StreamsResultSet readCurrent() {
- return new StreamsResultSet(constructQueue());
+ int batchSize = 0;
+ Queue<StreamsDatum> batch = Queues.newLinkedBlockingQueue();
+ try {
+ while (!this.data.isEmpty() && batchSize < DEFAULT_BATCH_SIZE) {
+ batch.add(this.data.take());
+ ++batchSize;
+ }
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+ System.out.println("******************\n**\tBatchSize="+batch.size()+"\n******************");
+ this.complete = batch.isEmpty() && this.data.isEmpty();
+ return new StreamsResultSet(batch);
}
@Override
@@ -63,12 +80,12 @@ public class NumericMessageProvider implements StreamsProvider {
@Override
public boolean isRunning() {
- return false;
+ return !this.complete;
}
@Override
public void prepare(Object configurationObject) {
-
+ this.data = constructQueue();
}
@Override
@@ -76,8 +93,8 @@ public class NumericMessageProvider implements StreamsProvider {
}
- private Queue<StreamsDatum> constructQueue() {
- Queue<StreamsDatum> datums = Queues.newArrayBlockingQueue(numMessages);
+ private BlockingQueue<StreamsDatum> constructQueue() {
+ BlockingQueue<StreamsDatum> datums = Queues.newArrayBlockingQueue(numMessages);
for(int i=0;i<numMessages;i++) {
datums.add(new StreamsDatum(i));
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/c2ade3ad/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/writer/DatumCounterWriter.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/writer/DatumCounterWriter.java b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/writer/DatumCounterWriter.java
index af86044..3c6c327 100644
--- a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/writer/DatumCounterWriter.java
+++ b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/test/writer/DatumCounterWriter.java
@@ -18,29 +18,84 @@
package org.apache.streams.local.test.writer;
+import com.google.common.collect.Lists;
import org.apache.streams.core.StreamsDatum;
import org.apache.streams.core.StreamsPersistWriter;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
/**
- * Created by rebanks on 2/18/14.
+ *
*/
public class DatumCounterWriter implements StreamsPersistWriter{
+ /**
+ * Set of all ids that have been claimed. Ensures all instances are assigned unique ids
+ */
+ public static Set<Integer> CLAIMED_ID = new HashSet<Integer>();
+ /**
+ * Random instance to generate ids
+ */
+ public static final Random RAND = new Random();
+ /**
+ * Set of instance ids that received data. Usefully for testing parrallelization is actually working.
+ */
+ public final static Set<Integer> SEEN_DATA = Collections.newSetFromMap(new ConcurrentHashMap<Integer, Boolean>());
+ /**
+ * The total count of data seen by a all instances of a processor.
+ */
+ public static final ConcurrentHashMap<String, AtomicLong> COUNTS = new ConcurrentHashMap<>();
+ /**
+ * The documents received
+ */
+ public static final ConcurrentHashMap<String, List<Object>> RECEIVED = new ConcurrentHashMap<>();
+
private int counter = 0;
+ private String writerId;
+ private Integer id;
+
+ public DatumCounterWriter(String writerId) {
+ this.writerId = writerId;
+ }
@Override
public void write(StreamsDatum entry) {
++this.counter;
+ SEEN_DATA.add(this.id);
+ synchronized (RECEIVED) {
+ List<Object> documents = RECEIVED.get(this.writerId);
+ if(documents == null) {
+ List<Object> docs = Lists.newLinkedList();
+ docs.add(entry.getDocument());
+ RECEIVED.put(this.writerId, docs);
+ } else {
+ documents.add(entry.getDocument());
+ }
+ }
}
@Override
public void prepare(Object configurationObject) {
-
+ synchronized (CLAIMED_ID) {
+ this.id = RAND.nextInt();
+ while(!CLAIMED_ID.add(this.id)) {
+ this.id = RAND.nextInt();
+ }
+ }
}
@Override
public void cleanUp() {
- System.out.println("clean up called");
+ synchronized (COUNTS) {
+ AtomicLong count = COUNTS.get(this.writerId);
+ if(count == null) {
+ COUNTS.put(this.writerId, new AtomicLong(this.counter));
+ } else {
+ count.addAndGet(this.counter);
+ }
+ }
}
public int getDatumsCounted() {
[10/11] git commit: STREAMS-191 | Changed addToOutgoingQueue to
reduce code complexity and to clarify code
Posted by re...@apache.org.
STREAMS-191 | Changed addToOutgoingQueue to reduce code complexity and to clarify code
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/7882f148
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/7882f148
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/7882f148
Branch: refs/heads/master
Commit: 7882f1480399289b1580235120fa7ee6bc7cd0ff
Parents: ae01621
Author: Ryan Ebanks <ry...@gmail.com>
Authored: Wed Oct 15 11:03:23 2014 -0500
Committer: Ryan Ebanks <ry...@gmail.com>
Committed: Wed Oct 15 11:03:23 2014 -0500
----------------------------------------------------------------------
.../streams/local/tasks/BaseStreamsTask.java | 33 +++++---------------
1 file changed, 7 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/7882f148/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java
index 902a2d7..6755d77 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BaseStreamsTask.java
@@ -99,36 +99,17 @@ public abstract class BaseStreamsTask implements StreamsTask {
outQueues.get(0).put(datum);
}
else {
- StreamsDatum newDatum = null;
- List<BlockingQueue<StreamsDatum>> failedQueues = Lists.newLinkedList();
- // TODO
- // Needs to be optimized better but workable now
- // Adds datums to queues that aren't full, then adds to full queues with blocking
- for(BlockingQueue<StreamsDatum> queue : this.outQueues) {
- try {
- newDatum = cloneStreamsDatum(datum);
- if(newDatum != null) {
- if(!queue.offer(newDatum, 500, TimeUnit.MILLISECONDS)) {
- failedQueues.add(queue);
+ List<BlockingQueue<StreamsDatum>> toOutput = Lists.newLinkedList(this.outQueues);
+ while(!toOutput.isEmpty()) {
+ for (BlockingQueue<StreamsDatum> queue : toOutput) {
+ StreamsDatum newDatum = cloneStreamsDatum(datum);
+ if (newDatum != null) {
+ if (queue.offer(newDatum, 500, TimeUnit.MILLISECONDS)) {
+ toOutput.remove(queue);
}
}
- } catch (RuntimeException e) {
- LOGGER.debug("Failed to add StreamsDatum to outgoing queue : {}", datum);
- LOGGER.error("Exception while offering StreamsDatum to outgoing queue: {}", e);
}
}
- for(BlockingQueue<StreamsDatum> queue : failedQueues) {
- try {
- newDatum = cloneStreamsDatum(datum);
- if(newDatum != null) {
- queue.put(newDatum);
- }
- } catch (RuntimeException e) {
- LOGGER.debug("Failed to add StreamsDatum to outgoing queue : {}", datum);
- LOGGER.error("Exception while offering StreamsDatum to outgoing queue: {}", e);
- }
- }
-
}
}
[04/11] git commit: Merge branch 'provider' into queues
Posted by re...@apache.org.
Merge branch 'provider' into queues
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/1b18eb65
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/1b18eb65
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/1b18eb65
Branch: refs/heads/master
Commit: 1b18eb6596b4afab3b798bbfd0dfa9928dbaae54
Parents: c2ade3a e74682e
Author: Ryan Ebanks <ry...@gmail.com>
Authored: Tue Oct 7 16:04:39 2014 -0500
Committer: Ryan Ebanks <ry...@gmail.com>
Committed: Tue Oct 7 16:04:39 2014 -0500
----------------------------------------------------------------------
.../apache/streams/local/tasks/StreamsProviderTask.java | 11 ++---------
1 file changed, 2 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/1b18eb65/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProviderTask.java
----------------------------------------------------------------------
[07/11] git commit: STREAMS-191 | Removed commented out code that was
in for refactoring purposes.
Posted by re...@apache.org.
STREAMS-191 | Removed commented out code that was in for refactoring purposes.
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/fd3ffe81
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/fd3ffe81
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/fd3ffe81
Branch: refs/heads/master
Commit: fd3ffe81cb4595f1c62d8a6e2e69dbe26db8ded9
Parents: 91a0f9a
Author: Ryan Ebanks <ry...@gmail.com>
Authored: Wed Oct 8 14:21:35 2014 -0500
Committer: Ryan Ebanks <ry...@gmail.com>
Committed: Wed Oct 8 14:21:35 2014 -0500
----------------------------------------------------------------------
.../local/tasks/StreamsPersistWriterTask.java | 23 ---------------
.../local/tasks/StreamsProcessorTask.java | 30 --------------------
2 files changed, 53 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/fd3ffe81/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 42b32e4..cab46b8 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
@@ -124,29 +124,6 @@ public class StreamsPersistWriterTask extends BaseStreamsTask implements DatumSt
LOGGER.debug("Received null StreamsDatum @ writer : {}", this.writer.getClass().getName());
}
}
-// StreamsDatum datum = this.inQueue.poll();
-// while(this.keepRunning.get()) {
-// if(datum != null) {
-// try {
-// this.writer.write(datum);
-// statusCounter.incrementStatus(DatumStatus.SUCCESS);
-// } catch (Exception e) {
-// LOGGER.error("Error writing to persist writer {}", this.writer.getClass().getSimpleName(), e);
-// this.keepRunning.set(false);
-// statusCounter.incrementStatus(DatumStatus.FAIL);
-// DatumUtils.addErrorToMetadata(datum, e, this.writer.getClass());
-// }
-// }
-// else {
-// try {
-// Thread.sleep(this.sleepTime);
-// } catch (InterruptedException e) {
-// LOGGER.warn("Thread interrupted in Writer task for {}",this.writer.getClass().getSimpleName(), e);
-// this.keepRunning.set(false);
-// }
-// }
-// datum = this.inQueue.poll();
-// }
} catch(Exception e) {
LOGGER.error("Failed to execute Persist Writer {}",this.writer.getClass().getSimpleName(), e);
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/fd3ffe81/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 a3d367f..ee69127 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
@@ -139,36 +139,6 @@ public class StreamsProcessorTask extends BaseStreamsTask implements DatumStatus
LOGGER.debug("Removed NULL datum from queue at processor : {}", this.processor.getClass().getName());
}
}
-
-// this.processor.prepare(this.streamConfig);
-// StreamsDatum datum = this.inQueue.poll();
-// while(this.keepRunning.get()) {
-// if(datum != null) {
-// try {
-// List<StreamsDatum> output = this.processor.process(datum);
-// if(output != null) {
-// for(StreamsDatum outDatum : output) {
-// super.addToOutgoingQueue(outDatum);
-// statusCounter.incrementStatus(DatumStatus.SUCCESS);
-// }
-// }
-// } catch (Throwable e) {
-// LOGGER.error("Throwable Streams Processor {}", e);
-// statusCounter.incrementStatus(DatumStatus.FAIL);
-// //Add the error to the metadata, but keep processing
-// DatumUtils.addErrorToMetadata(datum, e, this.processor.getClass());
-// }
-// }
-// else {
-// try {
-// Thread.sleep(this.sleepTime);
-// } catch (InterruptedException e) {
-// this.keepRunning.set(false);
-// }
-// }
-// datum = this.inQueue.poll();
-// }
-
} finally {
this.isRunning.set(false);
this.processor.cleanUp();
[09/11] git commit: STREAMS-191 | Merged apache master
Posted by re...@apache.org.
STREAMS-191 | Merged apache master
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/ae016213
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/ae016213
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/ae016213
Branch: refs/heads/master
Commit: ae0162135b6cf9ef277a6cf1d0b4b3f60df4cfcf
Parents: fd3ffe8 4704012
Author: Ryan Ebanks <ry...@gmail.com>
Authored: Mon Oct 13 11:01:39 2014 -0500
Committer: Ryan Ebanks <ry...@gmail.com>
Committed: Mon Oct 13 11:01:39 2014 -0500
----------------------------------------------------------------------
pom.xml | 9 +-
streams-config/pom.xml | 1 -
.../streams-persist-s3/pom.xml | 1 -
streams-contrib/streams-persist-hdfs/pom.xml | 13 +-
.../apache/streams/hdfs/HdfsConfigurator.java | 38 +-
.../streams/hdfs/WebHdfsPersistReader.java | 13 +-
.../streams/hdfs/WebHdfsPersistReaderTask.java | 5 +-
.../streams/hdfs/WebHdfsPersistWriter.java | 14 +-
.../apache/streams/hdfs/HdfsConfiguration.json | 6 +
.../streams/hdfs/HdfsWriterConfiguration.json | 5 +
.../hdfs/test/HdfsPersistConfigTest.java | 169 +++
streams-contrib/streams-persist-kafka/pom.xml | 1 -
streams-contrib/streams-persist-mongo/pom.xml | 1 -
.../streams-provider-datasift/pom.xml | 2 -
.../streams-provider-facebook/pom.xml | 1 -
.../gnip-edc-facebook/pom.xml | 1 -
.../gnip-edc-youtube/pom.xml | 3 +-
.../gnip-powertrack/pom.xml | 39 +-
.../google-gmail/pom.xml | 1 -
.../google-gplus/pom.xml | 1 -
.../streams-provider-instagram/pom.xml | 1 -
.../streams-provider-moreover/pom.xml | 3 +-
streams-contrib/streams-provider-rss/pom.xml | 50 +-
.../streams/rss/provider/RssStreamProvider.java | 169 ++-
.../rss/provider/RssStreamProviderTask.java | 214 +++-
.../provider/perpetual/RssFeedScheduler.java | 112 ++
.../rss/provider/RssStreamProviderTaskTest.java | 148 +++
.../rss/provider/RssStreamProviderTest.java | 103 ++
.../perpetual/RssFeedSchedulerTest.java | 101 ++
.../test/resources/test_rss_xml/economist1.xml | 233 ++++
.../test/resources/test_rss_xml/economist2.xml | 1069 ++++++++++++++++++
.../test_rss_xml/economistCombined.xml | 221 ++++
.../streams-provider-sysomos/pom.xml | 1 -
.../streams-provider-twitter/pom.xml | 1 -
.../provider/TwitterTimelineProvider.java | 10 +-
.../provider/TwitterTimelineProviderTask.java | 6 +-
streams-pojo/pom.xml | 1 -
.../streams/local/queues/ThroughputQueue.java | 289 +++--
.../local/queues/ThroughputQueueMXBean.java | 22 +-
.../queues/ThroughputQueueMulitThreadTest.java | 17 +
.../queues/ThroughputQueueSingleThreadTest.java | 90 +-
41 files changed, 2834 insertions(+), 351 deletions(-)
----------------------------------------------------------------------
[06/11] git commit: Added javadocs and set default queue size to 500
Posted by re...@apache.org.
Added javadocs and set default queue size to 500
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/91a0f9a6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/91a0f9a6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/91a0f9a6
Branch: refs/heads/master
Commit: 91a0f9a6799135619d2688699dcd71cfd3321138
Parents: 0d20773
Author: Ryan Ebanks <ry...@gmail.com>
Authored: Wed Oct 8 12:57:24 2014 -0500
Committer: Ryan Ebanks <ry...@gmail.com>
Committed: Wed Oct 8 12:57:24 2014 -0500
----------------------------------------------------------------------
.../streams/local/builders/LocalStreamBuilder.java | 15 +++++++++------
1 file changed, 9 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/91a0f9a6/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 4606aa7..57f3aa4 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
@@ -44,6 +44,7 @@ import java.util.concurrent.TimeUnit;
public class LocalStreamBuilder implements StreamBuilder {
private static final Logger LOGGER = org.slf4j.LoggerFactory.getLogger(LocalStreamBuilder.class);
+ private static final int DEFAULT_QUEUE_SIZE = 500;
public static final String TIMEOUT_KEY = "TIMEOUT";
private Map<String, StreamComponent> providers;
@@ -60,22 +61,23 @@ public class LocalStreamBuilder implements StreamBuilder {
private int maxQueueCapacity;
/**
- *
+ * Creates a local stream builder with no config object and default maximum internal queue size of 500
*/
public LocalStreamBuilder(){
- this(-1, null);
+ this(DEFAULT_QUEUE_SIZE, null);
}
/**
- *
+ * Creates a local stream builder with a config object and default maximum internal queue size of 500
* @param streamConfig
*/
public LocalStreamBuilder(Map<String, Object> streamConfig) {
- this(-1, streamConfig);
+ this(DEFAULT_QUEUE_SIZE, streamConfig);
}
/**
- *
+ * Creates a local stream builder with no config object. If maxQueueCapacity is less than 1 the queue is
+ * unbounded.
* @param maxQueueCapacity
*/
public LocalStreamBuilder(int maxQueueCapacity) {
@@ -83,7 +85,8 @@ public class LocalStreamBuilder implements StreamBuilder {
}
/**
- *
+ * Creates a local stream builder with a config object. If maxQueueCapacity is less than 1 the queue is
+ * unbounded.
* @param maxQueueCapacity
* @param streamConfig
*/
[02/11] git commit: Added blocking queues to test so that test will
compile
Posted by re...@apache.org.
Added blocking queues to test so that test will compile
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/3fad21a1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/3fad21a1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/3fad21a1
Branch: refs/heads/master
Commit: 3fad21a109289d4cd5b1e5444e2e0d8748154cbf
Parents: fb11add
Author: Ryan Ebanks <ry...@gmail.com>
Authored: Tue Oct 7 09:50:17 2014 -0500
Committer: Ryan Ebanks <ry...@gmail.com>
Committed: Tue Oct 7 09:50:17 2014 -0500
----------------------------------------------------------------------
.../streams/local/builders/StreamComponent.java | 6 +--
.../streams/local/tasks/StreamsMergeTask.java | 8 +++-
.../local/builders/LocalStreamBuilderTest.java | 4 +-
.../local/builders/ToyLocalBuilderExample.java | 5 +--
.../streams/local/tasks/BasicTasksTest.java | 44 ++++++++++----------
.../local/tasks/StreamsProviderTaskTest.java | 7 +---
6 files changed, 38 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3fad21a1/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java
index 5131227..8bccdf7 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java
@@ -189,13 +189,13 @@ public class StreamComponent {
if(this.numTasks > 1) {
task = new StreamsProcessorTask((StreamsProcessor)SerializationUtil.cloneBySerialization(this.processor));
task.addInputQueue(this.inQueue);
- for(Queue<StreamsDatum> q : this.outBound.values()) {
+ for(BlockingQueue<StreamsDatum> q : this.outBound.values()) {
task.addOutputQueue(q);
}
} else {
task = new StreamsProcessorTask(this.processor);
task.addInputQueue(this.inQueue);
- for(Queue<StreamsDatum> q : this.outBound.values()) {
+ for(BlockingQueue<StreamsDatum> q : this.outBound.values()) {
task.addOutputQueue(q);
}
}
@@ -226,7 +226,7 @@ public class StreamComponent {
if(timeout != 0) {
((StreamsProviderTask)task).setTimeout(timeout);
}
- for(Queue<StreamsDatum> q : this.outBound.values()) {
+ for(BlockingQueue<StreamsDatum> q : this.outBound.values()) {
task.addOutputQueue(q);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3fad21a1/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsMergeTask.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsMergeTask.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsMergeTask.java
index 4237be8..585e4dd 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsMergeTask.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsMergeTask.java
@@ -26,7 +26,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
/**
* NOT USED. When joins/partions are implemented, a similar pattern could be followed. Done only as basic proof
* of concept.
+ * NEEDS TO BE RE-WRITTEN
*/
+@Deprecated
public class StreamsMergeTask extends BaseStreamsTask {
private AtomicBoolean keepRunning;
@@ -62,7 +64,11 @@ public class StreamsMergeTask extends BaseStreamsTask {
while(this.keepRunning.get()) {
StreamsDatum datum = super.getNextDatum();
if(datum != null) {
- super.addToOutgoingQueue(datum);
+ try {
+ super.addToOutgoingQueue(datum);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
}
else {
try {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3fad21a1/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java
index c02f704..b082168 100644
--- a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java
+++ b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/LocalStreamBuilderTest.java
@@ -223,7 +223,7 @@ public class LocalStreamBuilderTest {
int timeout = 10000;
config.put(LocalStreamBuilder.TIMEOUT_KEY, timeout);
long start = System.currentTimeMillis();
- StreamBuilder builder = new LocalStreamBuilder(Queues.<StreamsDatum>newLinkedBlockingQueue(), config);
+ StreamBuilder builder = new LocalStreamBuilder(-1, config);
builder.newPerpetualStream("prov1", new EmptyResultSetProvider())
.addStreamsProcessor("proc1", new PassthroughDatumCounterProcessor(), 1, "prov1")
.addStreamsProcessor("proc2", new PassthroughDatumCounterProcessor(), 1, "proc1")
@@ -238,7 +238,7 @@ public class LocalStreamBuilderTest {
public void ensureShutdownWithBlockedQueue() throws InterruptedException {
ExecutorService service = Executors.newSingleThreadExecutor();
int before = Thread.activeCount();
- final StreamBuilder builder = new LocalStreamBuilder(Queues.<StreamsDatum>newLinkedBlockingQueue(1));
+ final StreamBuilder builder = new LocalStreamBuilder(1);
builder.newPerpetualStream("prov1", new NumericMessageProvider(30))
.addStreamsProcessor("proc1", new SlowProcessor(), 1, "prov1")
.addStreamsPersistWriter("w1", new SystemOutWriter(), 1, "proc1");
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3fad21a1/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/ToyLocalBuilderExample.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/ToyLocalBuilderExample.java b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/ToyLocalBuilderExample.java
index f2bccef..a77dfec 100644
--- a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/ToyLocalBuilderExample.java
+++ b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/builders/ToyLocalBuilderExample.java
@@ -19,12 +19,11 @@
package org.apache.streams.local.builders;
import org.apache.streams.core.StreamBuilder;
-import org.apache.streams.core.StreamsDatum;
import org.apache.streams.local.test.processors.DoNothingProcessor;
import org.apache.streams.local.test.providers.NumericMessageProvider;
import org.apache.streams.local.test.writer.DoNothingWriter;
-import java.util.concurrent.LinkedBlockingQueue;
+
/**
* Created by rebanks on 2/20/14.
@@ -36,7 +35,7 @@ public class ToyLocalBuilderExample {
* @param args
*/
public static void main(String[] args) {
- StreamBuilder builder = new LocalStreamBuilder(new LinkedBlockingQueue<StreamsDatum>());
+ StreamBuilder builder = new LocalStreamBuilder();
builder.newReadCurrentStream("prov", new NumericMessageProvider(1000000))
.addStreamsProcessor("proc", new DoNothingProcessor(), 100, "prov")
.addStreamsPersistWriter("writer", new DoNothingWriter(), 3, "proc");
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3fad21a1/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/BasicTasksTest.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/BasicTasksTest.java b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/BasicTasksTest.java
index 4ec4acc..c4ea4ee 100644
--- a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/BasicTasksTest.java
+++ b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/BasicTasksTest.java
@@ -19,21 +19,19 @@
package org.apache.streams.local.tasks;
import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.local.queues.ThroughputQueue;
import org.apache.streams.local.test.processors.PassthroughDatumCounterProcessor;
import org.apache.streams.local.test.providers.NumericMessageProvider;
import org.apache.streams.local.test.writer.DatumCounterWriter;
import org.junit.Test;
import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
import static org.junit.Assert.*;
/**
- * Created by rebanks on 2/18/14.
+ *
*/
public class BasicTasksTest {
@@ -44,9 +42,10 @@ public class BasicTasksTest {
int numMessages = 100;
NumericMessageProvider provider = new NumericMessageProvider(numMessages);
StreamsProviderTask task = new StreamsProviderTask(provider, false);
- Queue<StreamsDatum> outQueue = new ConcurrentLinkedQueue<StreamsDatum>();
+ BlockingQueue<StreamsDatum> outQueue = new LinkedBlockingQueue<>();
task.addOutputQueue(outQueue);
- Queue<StreamsDatum> inQueue = createInputQueue(numMessages);
+ //Test that adding input queues to providers is not valid
+ BlockingQueue<StreamsDatum> inQueue = createInputQueue(numMessages);
Exception exp = null;
try {
task.addInputQueue(inQueue);
@@ -54,6 +53,7 @@ public class BasicTasksTest {
exp = uoe;
}
assertNotNull(exp);
+
ExecutorService service = Executors.newFixedThreadPool(1);
service.submit(task);
int attempts = 0;
@@ -61,7 +61,7 @@ public class BasicTasksTest {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
- //Ignore
+ Thread.currentThread().interrupt();
}
++attempts;
if(attempts == 10) {
@@ -76,7 +76,7 @@ public class BasicTasksTest {
}
assertTrue("Task should have completed running in aloted time.", service.isTerminated());
} catch (InterruptedException e) {
- fail("Test Interupted.");
+ Thread.currentThread().interrupt();
};
}
@@ -85,8 +85,8 @@ public class BasicTasksTest {
int numMessages = 100;
PassthroughDatumCounterProcessor processor = new PassthroughDatumCounterProcessor();
StreamsProcessorTask task = new StreamsProcessorTask(processor);
- Queue<StreamsDatum> outQueue = new ConcurrentLinkedQueue<StreamsDatum>();
- Queue<StreamsDatum> inQueue = createInputQueue(numMessages);
+ BlockingQueue<StreamsDatum> outQueue = new LinkedBlockingQueue<>();
+ BlockingQueue<StreamsDatum> inQueue = createInputQueue(numMessages);
task.addOutputQueue(outQueue);
task.addInputQueue(inQueue);
assertEquals(numMessages, task.getInputQueues().get(0).size());
@@ -123,8 +123,8 @@ public class BasicTasksTest {
int numMessages = 100;
DatumCounterWriter writer = new DatumCounterWriter();
StreamsPersistWriterTask task = new StreamsPersistWriterTask(writer);
- Queue<StreamsDatum> outQueue = new ConcurrentLinkedQueue<StreamsDatum>();
- Queue<StreamsDatum> inQueue = createInputQueue(numMessages);
+ BlockingQueue<StreamsDatum> outQueue = new LinkedBlockingQueue<>();
+ BlockingQueue<StreamsDatum> inQueue = createInputQueue(numMessages);
Exception exp = null;
try {
@@ -168,7 +168,7 @@ public class BasicTasksTest {
int numMessages = 100;
int incoming = 5;
StreamsMergeTask task = new StreamsMergeTask();
- Queue<StreamsDatum> outQueue = new ConcurrentLinkedQueue<StreamsDatum>();
+ BlockingQueue<StreamsDatum> outQueue = new LinkedBlockingQueue<>();
task.addOutputQueue(outQueue);
for(int i=0; i < incoming; ++i) {
task.addInputQueue(createInputQueue(numMessages));
@@ -205,9 +205,9 @@ public class BasicTasksTest {
int numMessages = 100;
PassthroughDatumCounterProcessor processor = new PassthroughDatumCounterProcessor();
StreamsProcessorTask task = new StreamsProcessorTask(processor);
- Queue<StreamsDatum> outQueue1 = new ConcurrentLinkedQueue<StreamsDatum>();
- Queue<StreamsDatum> outQueue2 = new ConcurrentLinkedQueue<StreamsDatum>();
- Queue<StreamsDatum> inQueue = createInputQueue(numMessages);
+ BlockingQueue<StreamsDatum> outQueue1 = new LinkedBlockingQueue<>();
+ BlockingQueue<StreamsDatum> outQueue2 = new LinkedBlockingQueue<>();
+ BlockingQueue<StreamsDatum> inQueue = createInputQueue(numMessages);
task.addOutputQueue(outQueue1);
task.addOutputQueue(outQueue2);
task.addInputQueue(inQueue);
@@ -248,9 +248,9 @@ public class BasicTasksTest {
int numMessages = 1;
PassthroughDatumCounterProcessor processor = new PassthroughDatumCounterProcessor();
StreamsProcessorTask task = new StreamsProcessorTask(processor);
- Queue<StreamsDatum> outQueue1 = new ConcurrentLinkedQueue<StreamsDatum>();
- Queue<StreamsDatum> outQueue2 = new ConcurrentLinkedQueue<StreamsDatum>();
- Queue<StreamsDatum> inQueue = createInputQueue(numMessages);
+ BlockingQueue<StreamsDatum> outQueue1 = new LinkedBlockingQueue<>();
+ BlockingQueue<StreamsDatum> outQueue2 = new LinkedBlockingQueue<>();
+ BlockingQueue<StreamsDatum> inQueue = createInputQueue(numMessages);
task.addOutputQueue(outQueue1);
task.addOutputQueue(outQueue2);
task.addInputQueue(inQueue);
@@ -291,8 +291,8 @@ public class BasicTasksTest {
assertNotEquals(datum1, datum2);
}
- private Queue<StreamsDatum> createInputQueue(int numDatums) {
- Queue<StreamsDatum> queue = new ConcurrentLinkedQueue<StreamsDatum>();
+ private BlockingQueue<StreamsDatum> createInputQueue(int numDatums) {
+ BlockingQueue<StreamsDatum> queue = new LinkedBlockingQueue<>();
for(int i=0; i < numDatums; ++i) {
queue.add(new StreamsDatum(i));
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3fad21a1/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/StreamsProviderTaskTest.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/StreamsProviderTaskTest.java b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/StreamsProviderTaskTest.java
index 2cc4fed..856c357 100644
--- a/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/StreamsProviderTaskTest.java
+++ b/streams-runtimes/streams-runtime-local/src/test/java/org/apache/streams/local/tasks/StreamsProviderTaskTest.java
@@ -27,10 +27,7 @@ import org.junit.Before;
import org.junit.Test;
import java.util.Queue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.*;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.is;
@@ -67,7 +64,7 @@ public class StreamsProviderTaskTest {
@Test
public void flushes() {
- Queue<StreamsDatum> out = new LinkedBlockingQueue<>();
+ BlockingQueue<StreamsDatum> out = new LinkedBlockingQueue<>();
StreamsProviderTask task = new StreamsProviderTask(mockProvider, true);
when(mockProvider.isRunning()).thenReturn(true);
when(mockProvider.readCurrent()).thenReturn(new StreamsResultSet(getQueue(3)));