@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 testChoosePartitionWithKeyWithoutSequenceId() {
        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(-1L);
        Message<?> msg2 = mock(Message.class);
        when(msg2.hasKey()).thenReturn(true);
        when(msg2.getKey()).thenReturn(key2);
        when(msg1.getSequenceId()).thenReturn(-1L);

        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));
    }