diff --git a/local/twitter-userstream-elasticsearch/pom.xml b/local/twitter-userstream-elasticsearch/pom.xml
index 1a3074c..f438e77 100644
--- a/local/twitter-userstream-elasticsearch/pom.xml
+++ b/local/twitter-userstream-elasticsearch/pom.xml
@@ -81,6 +81,11 @@
         </dependency>
         <dependency>
             <groupId>org.apache.streams</groupId>
+            <artifactId>streams-filters</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.streams</groupId>
             <artifactId>streams-provider-twitter</artifactId>
             <version>${project.version}</version>
         </dependency>
@@ -240,4 +245,44 @@
             </plugin>
         </plugins>
     </build>
+
+    <profiles>
+        <profile>
+            <id>docker</id>
+            <build>
+                <plugins>
+                    <plugin>
+                        <!-- The Docker Maven plugin is used to create docker image with the fat jar -->
+                        <groupId>org.jolokia</groupId>
+                        <artifactId>docker-maven-plugin</artifactId>
+                        <version>0.11.0</version>
+                        <configuration>
+                            <images>
+
+                                <image>
+                                    <alias>${project.artifactId}</alias>
+                                    <name>${project.artifactId}:${project.version}</name>
+                                    <build>
+                                        <from>dockerfile/java:oracle-java8</from>
+                                        <assembly>
+                                            <basedir>/</basedir>
+                                            <descriptorRef>artifact</descriptorRef>
+                                        </assembly>
+                                        <!-- Default command for the build image -->
+                                    </build>
+
+                                </image>
+
+                            </images>
+                        </configuration>
+
+                    </plugin>
+
+                </plugins>
+            </build>
+            <activation>
+                <activeByDefault/>
+            </activation>
+        </profile>
+    </profiles>
 </project>
diff --git a/local/twitter-userstream-elasticsearch/src/main/java/org/apache/streams/twitter/example/TwitterUserstreamElasticsearch.java b/local/twitter-userstream-elasticsearch/src/main/java/org/apache/streams/twitter/example/TwitterUserstreamElasticsearch.java
index f24b1c4..2e7d2cd 100644
--- a/local/twitter-userstream-elasticsearch/src/main/java/org/apache/streams/twitter/example/TwitterUserstreamElasticsearch.java
+++ b/local/twitter-userstream-elasticsearch/src/main/java/org/apache/streams/twitter/example/TwitterUserstreamElasticsearch.java
@@ -18,45 +18,26 @@
 
 package org.apache.streams.twitter.example;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
-import com.typesafe.config.Config;
+import com.google.common.collect.Sets;
 import org.apache.streams.config.ComponentConfigurator;
 import org.apache.streams.config.StreamsConfigurator;
-import org.apache.streams.core.StreamsDatum;
-import org.apache.streams.core.StreamsProcessor;
-import org.apache.streams.data.util.ActivityUtil;
-import org.apache.streams.elasticsearch.ElasticsearchConfigurator;
+import org.apache.streams.converter.ActivityConverterProcessor;
 import org.apache.streams.elasticsearch.ElasticsearchPersistDeleter;
 import org.apache.streams.elasticsearch.ElasticsearchPersistWriter;
 import org.apache.streams.elasticsearch.ElasticsearchWriterConfiguration;
 import org.apache.streams.example.twitter.TwitterUserstreamElasticsearchConfiguration;
-import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.filters.VerbDefinitionDropFilter;
+import org.apache.streams.filters.VerbDefinitionKeepFilter;
 import org.apache.streams.local.builders.LocalStreamBuilder;
 import org.apache.streams.core.StreamBuilder;
-import org.apache.streams.pojo.json.Activity;
-import org.apache.streams.pojo.json.ActivityObject;
-import org.apache.streams.pojo.json.Actor;
-import org.apache.streams.pojo.json.Delete;
-import org.apache.streams.pojo.json.Follow;
-import org.apache.streams.pojo.json.Page;
 import org.apache.streams.twitter.TwitterStreamConfiguration;
-import org.apache.streams.twitter.pojo.FriendList;
-import org.apache.streams.twitter.pojo.UserstreamEvent;
-import org.apache.streams.twitter.processor.TwitterTypeConverter;
-import org.apache.streams.twitter.provider.TwitterConfigurator;
 import org.apache.streams.twitter.provider.TwitterStreamProvider;
-import org.apache.streams.twitter.serializer.StreamsTwitterMapper;
-import org.apache.streams.twitter.serializer.util.TwitterActivityUtil;
+import org.apache.streams.verbs.VerbDefinition;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.List;
 import java.util.Map;
-import java.util.concurrent.LinkedBlockingQueue;
 
 /**
  * Example stream that populates elasticsearch with activities from twitter userstream in real-time
@@ -67,6 +48,8 @@
 
     private final static Logger LOGGER = LoggerFactory.getLogger(TwitterUserstreamElasticsearch.class);
 
+    private static VerbDefinition deleteVerbDefinition = new VerbDefinition().withValue("post");
+
     TwitterUserstreamElasticsearchConfiguration config;
 
     public TwitterUserstreamElasticsearch() {
@@ -94,67 +77,25 @@
         ElasticsearchWriterConfiguration elasticsearchWriterConfiguration = config.getElasticsearch();
 
         TwitterStreamProvider stream = new TwitterStreamProvider(twitterStreamConfiguration);
-        TwitterTypeConverter converter = new TwitterTypeConverter(ObjectNode.class, Activity.class);
+        ActivityConverterProcessor converter = new ActivityConverterProcessor();
+        VerbDefinitionDropFilter noDeletesProcessor = new VerbDefinitionDropFilter(Sets.newHashSet(deleteVerbDefinition));
         ElasticsearchPersistWriter writer = new ElasticsearchPersistWriter(elasticsearchWriterConfiguration);
-        DeleteOnlyProcessor deleteOnlyProcessor = new DeleteOnlyProcessor();
-        NoDeletesProcessor noDeletesProcessor = new NoDeletesProcessor();
+        VerbDefinitionKeepFilter deleteOnlyProcessor = new VerbDefinitionKeepFilter(Sets.newHashSet(deleteVerbDefinition));
         ElasticsearchPersistDeleter deleter = new ElasticsearchPersistDeleter(elasticsearchWriterConfiguration);
 
         Map<String, Object> streamConfig = Maps.newHashMap();
         streamConfig.put(LocalStreamBuilder.TIMEOUT_KEY, 12 * 60 * 1000);
-        StreamBuilder builder = new LocalStreamBuilder(25);
+        StreamBuilder builder = new LocalStreamBuilder(25, streamConfig);
 
         builder.newPerpetualStream(TwitterStreamProvider.STREAMS_ID, stream);
         builder.addStreamsProcessor("converter", converter, 2, TwitterStreamProvider.STREAMS_ID);
         builder.addStreamsProcessor("NoDeletesProcessor", noDeletesProcessor, 1, "converter");
         builder.addStreamsPersistWriter(ElasticsearchPersistWriter.STREAMS_ID, writer, 1, "NoDeletesProcessor");
         builder.addStreamsProcessor("DeleteOnlyProcessor", deleteOnlyProcessor, 1, "converter");
-        builder.addStreamsPersistWriter(ElasticsearchPersistDeleter.STREAMS_ID, deleter, 1, "DeleteOnlyProcessor");
+        builder.addStreamsPersistWriter("deleter", deleter, 1, "DeleteOnlyProcessor");
 
         builder.start();
 
     }
 
-    private class DeleteOnlyProcessor implements StreamsProcessor
-    {
-        String delete = new Delete().getVerb();
-
-        @Override
-        public void prepare(Object configurationObject) {}
-
-        @Override
-        public void cleanUp() {}
-
-        @Override
-        public List<StreamsDatum> process(StreamsDatum entry) {
-            Preconditions.checkArgument(entry.getDocument() instanceof Activity);
-            Activity activity = (Activity) entry.getDocument();
-            if( activity.getVerb().equals(delete))
-                return Lists.newArrayList(entry);
-            else
-                return Lists.newArrayList();
-        }
-    }
-
-    private class NoDeletesProcessor implements StreamsProcessor
-    {
-        String delete = new Delete().getVerb();
-
-        @Override
-        public void prepare(Object configurationObject) {}
-
-        @Override
-        public void cleanUp() {}
-
-        @Override
-        public List<StreamsDatum> process(StreamsDatum entry) {
-            Preconditions.checkArgument(entry.getDocument() instanceof Activity);
-            Activity activity = (Activity) entry.getDocument();
-            if( activity.getVerb().equals(delete))
-                return Lists.newArrayList();
-            else
-                return Lists.newArrayList(entry);
-        }
-    }
-
 }
diff --git a/local/twitter-userstream-elasticsearch/src/main/resources/TwitterUserstreamElasticsearch.dot b/local/twitter-userstream-elasticsearch/src/main/resources/TwitterUserstreamElasticsearch.dot
index 5302bf9..62aa637 100644
--- a/local/twitter-userstream-elasticsearch/src/main/resources/TwitterUserstreamElasticsearch.dot
+++ b/local/twitter-userstream-elasticsearch/src/main/resources/TwitterUserstreamElasticsearch.dot
@@ -5,8 +5,8 @@
 
   //processors
   TwitterTypeConverter [label="TwitterTypeConverter",shape=box,URL="https://github.com/apache/incubator-streams/blob/master/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterTypeConverter.java"];
-  DeleteOnlyProcessor  [label="DeleteOnlyProcessor ",shape=box,URL="https://github.com/apache/incubator-streams/blob/master/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterTypeConverter.java"];
-  NoDeletesProcessor  [label="NoDeletes",shape=box,URL="https://github.com/apache/incubator-streams/blob/master/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterTypeConverter.java"];
+  DeleteOnlyProcessor [label="VerbDefinitionKeepFilter (verb:post)",shape=box,URL="https://github.com/apache/incubator-streams/blob/master/streams-components/streams-filters/src/main/java/org/apache/streams/filters/VerbDefinitionKeepFilter.java"];
+  NoDeletesProcessor  [label="VerbDefinitionDropFilter (verb:post)",shape=box,URL="https://github.com/apache/incubator-streams/blob/master/streams-components/streams-filters/src/main/java/org/apache/streams/filters/VerbDefinitionDropFilter.java"];
    
   //persisters
   ElasticsearchPersistWriter [label="ElasticsearchPersistWriter",shape=ellipse,URL="https://github.com/apache/incubator-streams/blob/master/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java"];