diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/DefaultCommModule.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/DefaultCommModule.java
index d7c8cee..59409a7 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/DefaultCommModule.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/DefaultCommModule.java
@@ -47,7 +47,8 @@
 import com.google.inject.name.Names;
 
 /**
- * Default configuration module for the communication layer. Parameterizable through a configuration file.
+ * Default configuration module for the communication layer. Parameterizable / overridable through custom modules and /
+ * or configuration file (for string literals).
  * 
  */
 public class DefaultCommModule extends AbstractModule {
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java
index 936d225..673d0bb 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java
@@ -28,6 +28,7 @@
 import org.apache.s4.base.KeyFinder;
 import org.apache.s4.base.SerializerDeserializer;
 import org.apache.s4.comm.serialize.KryoSerDeser;
+import org.apache.s4.comm.topology.Cluster;
 import org.apache.s4.comm.topology.RemoteStreams;
 import org.apache.s4.core.ft.CheckpointingFramework;
 import org.apache.s4.core.window.AbstractSlidingWindowPE;
@@ -41,7 +42,7 @@
 
 /**
  * Container base class to hold all processing elements.
- *
+ * 
  * It is also where one defines the application graph: PE prototypes, internal streams, input and output streams.
  */
 public abstract class App {
@@ -57,7 +58,7 @@
     final private List<Streamable<Event>> streams = new ArrayList<Streamable<Event>>();
 
     /* Pes indexed by name. */
-    Map<String, ProcessingElement> peByName = Maps.newHashMap();
+    final Map<String, ProcessingElement> peByName = Maps.newHashMap();
 
     private ClockType clockType = ClockType.WALL_CLOCK;
     private int id = -1;
@@ -68,17 +69,20 @@
     private Receiver receiver;
 
     @Inject
-    RemoteSenders remoteSenders;
+    private Cluster cluster;
 
     @Inject
-    Hasher hasher;
+    private RemoteSenders remoteSenders;
 
     @Inject
-    RemoteStreams remoteStreams;
+    private Hasher hasher;
+
+    @Inject
+    private RemoteStreams remoteStreams;
 
     @Inject
     @Named("s4.cluster.name")
-    String clusterName;
+    private String clusterName;
 
     // default is NoOpCheckpointingFramework
     @Inject
@@ -256,6 +260,25 @@
     }
 
     /**
+     * 
+     * Returns the id of the partition assigned to the current node.
+     * 
+     * NOTE: This method will block until the current node gets assigned a partition
+     * 
+     */
+    public int getPartitionId() {
+        return getReceiver().getPartitionId();
+    }
+
+    /**
+     * 
+     * Returns the total number of partitions of the cluster this nodes belongs to.
+     */
+    public int getPartitionCount() {
+        return cluster.getPhysicalCluster().getPartitionCount();
+    }
+
+    /**
      * @return the sender object
      */
     public Sender getSender() {
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/DefaultCoreModule.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/DefaultCoreModule.java
index 5701640..0c42f48 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/DefaultCoreModule.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/DefaultCoreModule.java
@@ -41,8 +41,7 @@
 import com.google.inject.name.Names;
 
 /**
- * Temporary module allowing assignment from ZK, communication through Netty, and distributed deployment management,
- * until we have a better way to customize node configuration
+ * Default module allowing assignment from ZK, communication through Netty, and distributed deployment management
  * 
  */
 public class DefaultCoreModule extends AbstractModule {
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/Receiver.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/Receiver.java
index 6c0b19c..68840c0 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/Receiver.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/Receiver.java
@@ -67,7 +67,7 @@
         streams = new MapMaker().makeMap();
     }
 
-    int getPartition() {
+    int getPartitionId() {
         return listener.getPartitionId();
     }
 
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/Sender.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/Sender.java
index 5b0b03d..b5dcf6e 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/Sender.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/Sender.java
@@ -24,6 +24,8 @@
 import org.apache.s4.base.Hasher;
 import org.apache.s4.base.SerializerDeserializer;
 import org.apache.s4.comm.topology.Assignment;
+import org.apache.s4.comm.topology.Cluster;
+import org.apache.s4.comm.topology.ClusterChangeListener;
 import org.apache.s4.comm.topology.ClusterNode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -38,16 +40,17 @@
  * Details on how the cluster is partitioned and how events are serialized and transmitted to its destination are hidden
  * from the application developer.
  */
-public class Sender {
+public class Sender implements ClusterChangeListener {
 
     private static Logger logger = LoggerFactory.getLogger(Sender.class);
 
     final private Emitter emitter;
     final private SerializerDeserializer serDeser;
     final private Hasher hasher;
+    final private Cluster cluster;
 
-    Assignment assignment;
-    private int localPartitionId = -1;
+    final private Assignment assignment;
+    private volatile int localPartitionId = -1;
 
     /**
      * 
@@ -59,11 +62,14 @@
      *            a hashing function to map keys to partition IDs.
      */
     @Inject
-    public Sender(Emitter emitter, SerializerDeserializer serDeser, Hasher hasher, Assignment assignment) {
+    public Sender(Emitter emitter, SerializerDeserializer serDeser, Hasher hasher, Assignment assignment,
+            Cluster cluster) {
         this.emitter = emitter;
         this.serDeser = serDeser;
         this.hasher = hasher;
         this.assignment = assignment;
+        this.cluster = cluster;
+        this.cluster.addListener(this);
     }
 
     @Inject
@@ -121,4 +127,9 @@
         }
     }
 
+    @Override
+    public void onChange() {
+        resolveLocalPartitionId();
+    }
+
 }
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/MockCoreModule.java b/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/MockCoreModule.java
index 71bae7a..6a60ec5 100644
--- a/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/MockCoreModule.java
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/MockCoreModule.java
@@ -20,6 +20,8 @@
 
 import org.apache.s4.base.Emitter;
 import org.apache.s4.base.Listener;
+import org.apache.s4.comm.topology.Cluster;
+import org.apache.s4.comm.topology.PhysicalCluster;
 import org.apache.s4.core.Receiver;
 import org.apache.s4.deploy.DeploymentManager;
 import org.apache.s4.deploy.NoOpDeploymentManager;
@@ -47,5 +49,8 @@
         bind(Emitter.class).toInstance(Mockito.mock(Emitter.class));
         bind(Listener.class).toInstance(Mockito.mock(Listener.class));
         bind(Receiver.class).toInstance(Mockito.mock(Receiver.class));
+        Cluster clusterMock = Mockito.mock(Cluster.class);
+        Mockito.when(clusterMock.getPhysicalCluster()).thenReturn(new PhysicalCluster(1));
+        bind(Cluster.class).toInstance(clusterMock);
     }
 }
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCounterPE.java b/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCounterPE.java
index c9d8635..a124db3 100644
--- a/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCounterPE.java
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCounterPE.java
@@ -23,39 +23,45 @@
 import org.apache.s4.core.Stream;
 
 public class WordCounterPE extends ProcessingElement {
-    
+
     int wordCounter;
     transient Stream<WordCountEvent> wordClassifierStream;
 
-    private WordCounterPE() {}
-    
+    private WordCounterPE() {
+    }
+
     public WordCounterPE(App app) {
         super(app);
     }
-    
+
     public void setWordClassifierStream(Stream<WordCountEvent> stream) {
         this.wordClassifierStream = stream;
     }
 
-    public void onEvent(WordSeenEvent event) { 
+    public void onEvent(WordSeenEvent event) {
+
         wordCounter++;
         System.out.println("seen word " + event.getWord());
-        // NOTE: it seems the id is the key for now...     
+        // NOTE: it seems the id is the key for now...
         wordClassifierStream.put(new WordCountEvent(getId(), wordCounter));
+
+        // add some tests for partition count and id
+        if (!((getApp().getPartitionCount() == 1) && (getApp().getPartitionId() == 0))) {
+            throw new RuntimeException("Invalid partitioning");
+        }
+
     }
 
     @Override
     protected void onCreate() {
         // TODO Auto-generated method stub
-        
+
     }
 
     @Override
     protected void onRemove() {
         // TODO Auto-generated method stub
-        
-    }
 
-   
+    }
 
 }