public StickyPartitionCache() {
        this.indexCache = new ConcurrentHashMap<>();
    }
@Test
    public void testStickyPartitionCache() {
        List<PartitionInfo> allPartitions = asList(new PartitionInfo(TOPIC_A, 0, NODES[0], NODES, NODES),
            new PartitionInfo(TOPIC_A, 1, NODES[1], NODES, NODES),
            new PartitionInfo(TOPIC_A, 2, NODES[2], NODES, NODES),
            new PartitionInfo(TOPIC_B, 0, NODES[0], NODES, NODES)
        );
        Cluster testCluster = new Cluster("clusterId", asList(NODES), allPartitions,
            Collections.emptySet(), Collections.emptySet());
        StickyPartitionCache stickyPartitionCache = new StickyPartitionCache();

        int partA = stickyPartitionCache.partition(TOPIC_A, testCluster);
        assertEquals(partA, stickyPartitionCache.partition(TOPIC_A, testCluster));

        int partB = stickyPartitionCache.partition(TOPIC_B, testCluster);
        assertEquals(partB, stickyPartitionCache.partition(TOPIC_B, testCluster));

        int changedPartA = stickyPartitionCache.nextPartition(TOPIC_A, testCluster, partA);
        assertEquals(changedPartA, stickyPartitionCache.partition(TOPIC_A, testCluster));
        assertNotEquals(partA, changedPartA);
        int changedPartA2 = stickyPartitionCache.partition(TOPIC_A, testCluster);
        assertEquals(changedPartA2, changedPartA);

        // We do not want to change partitions because the previous partition does not match the current sticky one.
        int changedPartA3 = stickyPartitionCache.nextPartition(TOPIC_A, testCluster, partA);
        assertEquals(changedPartA3, changedPartA2);

        // Check that we can still use the partitioner when there is only one partition
        int changedPartB = stickyPartitionCache.nextPartition(TOPIC_B, testCluster, partB);
        assertEquals(changedPartB, stickyPartitionCache.partition(TOPIC_B, testCluster));
    }