@Override
    public int choosePartition(Message msg, TopicMetadata metadata) {
        // if key is specified, we should use key as routing;
        // if key is not specified and no sequence id is provided, not an effectively-once publish, use the default
        // round-robin routing.
        if (msg.hasKey() || msg.getSequenceId() < 0) {
            // TODO: the message key routing is problematic at this moment.
            //       https://github.com/apache/pulsar/pull/1029 is fixing that.
            return super.choosePartition(msg, metadata);
        }
        // if there is no key and sequence id is provided, it is an effectively-once publish, we need to ensure
        // for a given message it always go to one partition, so we use sequence id to do a deterministic routing.
        return (int) (msg.getSequenceId() % metadata.numPartitions());
    }
@Test
    public void testChoosePartitionWithoutKeySequenceId() {
        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++) {
            Message<?> msg = mock(Message.class);
            when(msg.hasKey()).thenReturn(false);
            when(msg.getKey()).thenReturn(null);
            when(msg.getSequenceId()).thenReturn((long) (2 * i));
            assertEquals((2 * i) % 5, router.choosePartition(msg, topicMetadata));
        }
    }