@Override
    public int choosePartition(Message<?> msg, TopicMetadata topicMetadata) {
        // If the message has a key, it supersedes the round robin routing policy
        if (msg.hasKey()) {
            return signSafeMod(hash.makeHash(msg.getKey()), topicMetadata.numPartitions());
        }

        if (isBatchingEnabled) { // if batching is enabled, choose partition on `partitionSwitchMs` boundary.
            long currentMs = clock.millis();
            return signSafeMod(currentMs / partitionSwitchMs + startPtnIdx, topicMetadata.numPartitions());
        } else {
            return signSafeMod(PARTITION_INDEX_UPDATER.getAndIncrement(this), topicMetadata.numPartitions());
        }
    }
@Test
    public void testChoosePartitionWithNegativeTime() {
        Message<?> msg = mock(Message.class);
        when(msg.getKey()).thenReturn(null);

        // Fake clock, simulate timestamp that resolves into a negative Integer value
        Clock clock = mock(Clock.class);
        when(clock.millis()).thenReturn((long) Integer.MAX_VALUE);

        RoundRobinPartitionMessageRouterImpl router = new RoundRobinPartitionMessageRouterImpl(
                HashingScheme.JavaStringHash, 3, true, 5, clock);

        int idx = router.choosePartition(msg, new TopicMetadataImpl(5));
        assertTrue(idx >= 0);
        assertTrue(idx < 5);
    }