Hi Matthieu: We can discuss this on the tuesday meetings but what would be great is to have support built into the RemoteStream to register a key. This way in the pattern: InputAdapter -> PartitionerPE -> CompPE1..ComPEn ->OutputAdapter. The OutputAdapter is a PE but let's discuss this on the call as well, it doesn't have to . - Shailendra
On Fri, Jun 8, 2012 at 8:50 AM, <[email protected]> wrote: > Updated Branches: > refs/heads/S4-22 6808e9ffc -> c34742d87 > > > by default, remote senders send keyless events in round-robin fashion > (rather than broadcast) > > > Project: http://git-wip-us.apache.org/repos/asf/incubator-s4/repo > Commit: http://git-wip-us.apache.org/repos/asf/incubator-s4/commit/c34742d8 > Tree: http://git-wip-us.apache.org/repos/asf/incubator-s4/tree/c34742d8 > Diff: http://git-wip-us.apache.org/repos/asf/incubator-s4/diff/c34742d8 > > Branch: refs/heads/S4-22 > Commit: c34742d8711b439f8f17c1d9417e62468ed59cb5 > Parents: 6808e9f > Author: Matthieu Morel <[email protected]> > Authored: Fri Jun 8 17:36:23 2012 +0200 > Committer: Matthieu Morel <[email protected]> > Committed: Fri Jun 8 17:36:23 2012 +0200 > > ---------------------------------------------------------------------- > .../main/java/org/apache/s4/core/RemoteSender.java | 6 +++--- > 1 files changed, 3 insertions(+), 3 deletions(-) > ---------------------------------------------------------------------- > > > http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/c34742d8/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSender.java > ---------------------------------------------------------------------- > diff --git > a/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSender.java > b/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSender.java > index cdb0b01..cbe037e 100644 > --- a/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSender.java > +++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSender.java > @@ -8,6 +8,7 @@ public class RemoteSender { > > final private Emitter emitter; > final private Hasher hasher; > + int targetPartition = 0; > > public RemoteSender(Emitter emitter, Hasher hasher) { > super(); > @@ -17,9 +18,8 @@ public class RemoteSender { > > public void send(String hashKey, EventMessage eventMessage) { > if (hashKey == null) { > - for (int i = 0; i < emitter.getPartitionCount(); i++) { > - emitter.send(i, eventMessage); > - } > + // round robin by default > + emitter.send(Math.abs(targetPartition++ % > emitter.getPartitionCount()), eventMessage); > } else { > int partition = (int) (hasher.hash(hashKey) % > emitter.getPartitionCount()); > emitter.send(partition, eventMessage); >
