@Test public void testChoosePartitionWithoutKeyWithoutSequenceId() { Message<?> msg = mock(Message.class); when(msg.hasKey()).thenReturn(false); when(msg.getKey()).thenReturn(null); when(msg.getSequenceId()).thenReturn(-1L); TopicMetadata topicMetadata = mock(TopicMetadata.class); when(topicMetadata.numPartitions()).thenReturn(5); Clock clock = mock(Clock.class); FunctionResultRouter router = new FunctionResultRouter(0, clock); for (int i = 0; i < 10; i++) { when(clock.millis()).thenReturn(123450L + i); assertEquals(i % 5, router.choosePartition(msg, topicMetadata)); } }
@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);  }