@Test public void testChoosePartitionWithKeySequenceId() { String key1 = "key1"; String key2 = "key2"; Message<?> msg1 = mock(Message.class); when(msg1.hasKey()).thenReturn(true); when(msg1.getKey()).thenReturn(key1); when(msg1.getSequenceId()).thenReturn((long) ((key1.hashCode() % 100) + 1)); Message<?> msg2 = mock(Message.class); when(msg2.hasKey()).thenReturn(true); when(msg2.getKey()).thenReturn(key2); when(msg1.getSequenceId()).thenReturn((long) ((key2.hashCode() % 100) + 1)); Clock clock = mock(Clock.class); FunctionResultRouter router = new FunctionResultRouter(0, clock); TopicMetadata metadata = mock(TopicMetadata.class); when(metadata.numPartitions()).thenReturn(100); assertEquals(hash.makeHash(key1) % 100, router.choosePartition(msg1, metadata)); assertEquals(hash.makeHash(key2) % 100, router.choosePartition(msg2, metadata)); }
@Override public int choosePartition(Message msg, TopicMetadata metadata) { if (msg.hasKey() || msg.getSequenceId() < 0) { return super.choosePartition(msg, metadata); } return (int) (msg.getSequenceId() % metadata.numPartitions()); }
FunctionResultRouter extends RoundRobinPartitionMessageRouterImpl { @Override public int choosePartition(Message msg, TopicMetadata metadata) { if (msg.hasKey() || msg.getSequenceId() < 0) { return super.choosePartition(msg, metadata); } return (int) (msg.getSequenceId() % metadata.numPartitions()); } }
FunctionResultRouter extends RoundRobinPartitionMessageRouterImpl { @Override public int choosePartition(Message msg, TopicMetadata metadata) { if (msg.hasKey() || msg.getSequenceId() < 0) { return super.choosePartition(msg, metadata); } return (int) (msg.getSequenceId() % metadata.numPartitions()); } FunctionResultRouter(); @VisibleForTesting  FunctionResultRouter(int startPtnIdx, Clock clock); }
FunctionResultRouter extends RoundRobinPartitionMessageRouterImpl { @Override public int choosePartition(Message msg, TopicMetadata metadata) { if (msg.hasKey() || msg.getSequenceId() < 0) { return super.choosePartition(msg, metadata); } return (int) (msg.getSequenceId() % metadata.numPartitions()); } FunctionResultRouter(); @VisibleForTesting  FunctionResultRouter(int startPtnIdx, Clock clock); static FunctionResultRouter of(); @Override int choosePartition(Message msg, TopicMetadata metadata); }
FunctionResultRouter extends RoundRobinPartitionMessageRouterImpl { @Override public int choosePartition(Message msg, TopicMetadata metadata) { if (msg.hasKey() || msg.getSequenceId() < 0) { return super.choosePartition(msg, metadata); } return (int) (msg.getSequenceId() % metadata.numPartitions()); } FunctionResultRouter(); @VisibleForTesting  FunctionResultRouter(int startPtnIdx, Clock clock); static FunctionResultRouter of(); @Override int choosePartition(Message msg, TopicMetadata metadata);  }