Repository: apache/kafka
File: core/src/test/java/kafka/server/share/SharePartitionManagerTest.java (Java)

Code before:
        assertThrows(InvalidRequestException.class, () -> sharePartitionManager.newContext("grp", List.of(tp0, tp1), List.of(),
            new ShareRequestMetadata(Uuid.randomUuid(), ShareRequestMetadata.INITIAL_EPOCH), true, CONNECTION_ID));
    }

    @Test
    public void testNewContextThrowsErrorWhenShareSessionCacheIsFullOnInitialEpoch() {
        // Define a cache with max size 1
        ShareSessionCache cache = new ShareSessionCache(1);
        sharePartitionManager = SharePartitionManagerBuilder.builder()
            .withCache(cache)
            .build();

        Uuid tpId0 = Uuid.randomUuid();
        TopicIdPartition tp0 = new TopicIdPartition(tpId0, new TopicPartition("foo", 0));
        TopicIdPartition tp1 = new TopicIdPartition(tpId0, new TopicPartition("foo", 1));

        String groupId = "grp";
        Uuid memberId1 = Uuid.randomUuid();
        Uuid memberId2 = Uuid.randomUuid();

        // Create a new share session with an initial share fetch request
        List<TopicIdPartition> reqData = List.of(tp0, tp1);

        ShareRequestMetadata reqMetadata1 = new ShareRequestMetadata(memberId1, ShareRequestMetadata.INITIAL_EPOCH);
        ShareFetchContext context1 = sharePartitionManager.newContext(groupId, reqData, EMPTY_PART_LIST, reqMetadata1, false, CONNECTION_ID);
        assertInstanceOf(ShareSessionContext.class, context1);
        assertFalse(((ShareSessionContext) context1).isSubsequent());

        // Trying to create a new share session, but since cache is already full, it should throw an exception
        ShareRequestMetadata reqMetadata2 = new ShareRequestMetadata(memberId2, ShareRequestMetadata.INITIAL_EPOCH);
        assertThrows(ShareSessionLimitReachedException.class, () -> sharePartitionManager.newContext("grp", reqData, EMPTY_PART_LIST,
            reqMetadata2, false, "id-2"));
    }

    @Test
    public void testNewContextExistingSessionNewRequestWithInitialEpoch() {
        ShareSessionCache cache = new ShareSessionCache(10);
        sharePartitionManager = SharePartitionManagerBuilder.builder()
            .withCache(cache)
            .build();

        Uuid tpId0 = Uuid.randomUuid();
        TopicIdPartition tp0 = new TopicIdPartition(tpId0, new TopicPartition("foo", 0));
        TopicIdPartition tp1 = new TopicIdPartition(tpId0, new TopicPartition("foo", 1));

        String groupId = "grp";
        Uuid memberId = Uuid.randomUuid();
        List<TopicIdPartition> reqData = List.of(tp0, tp1);

        ShareRequestMetadata reqMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH);


Review comment (suggestion):
In some cases in this test, we use `EMPTY_PART_LIST` which is `List.of()`. In some cases we do not. I would prefer to see them used consistently. Either use `EMPTY_PART_LIST` in every situation where an empty partition list is being provided, or remove `EMPTY_PART_LIST` entirely and use `List.of()` instead.

The test cases added by this PR are inconsistent in this regard. It would be good to be consistent across the entire source file too.

Code after:

        assertThrows(InvalidRequestException.class, () -> sharePartitionManager.newContext("grp", List.of(tp0, tp1), EMPTY_PART_LIST,
            new ShareRequestMetadata(Uuid.randomUuid(), ShareRequestMetadata.INITIAL_EPOCH), true, CONNECTION_ID));
    }

    @Test
    public void testNewContextThrowsErrorWhenShareSessionCacheIsFullOnInitialEpoch() {
        // Define a cache with max size 1
        ShareSessionCache cache = new ShareSessionCache(1);
        sharePartitionManager = SharePartitionManagerBuilder.builder()
            .withCache(cache)
            .build();

        Uuid tpId0 = Uuid.randomUuid();
        TopicIdPartition tp0 = new TopicIdPartition(tpId0, new TopicPartition("foo", 0));
        TopicIdPartition tp1 = new TopicIdPartition(tpId0, new TopicPartition("foo", 1));

        String groupId = "grp";
        Uuid memberId1 = Uuid.randomUuid();
        Uuid memberId2 = Uuid.randomUuid();

        // Create a new share session with an initial share fetch request
        List<TopicIdPartition> reqData = List.of(tp0, tp1);

        ShareRequestMetadata reqMetadata1 = new ShareRequestMetadata(memberId1, ShareRequestMetadata.INITIAL_EPOCH);
        ShareFetchContext context1 = sharePartitionManager.newContext(groupId, reqData, EMPTY_PART_LIST, reqMetadata1, false, CONNECTION_ID);
        assertInstanceOf(ShareSessionContext.class, context1);
        assertFalse(((ShareSessionContext) context1).isSubsequent());

        // Trying to create a new share session, but since cache is already full, it should throw an exception
        ShareRequestMetadata reqMetadata2 = new ShareRequestMetadata(memberId2, ShareRequestMetadata.INITIAL_EPOCH);
        assertThrows(ShareSessionLimitReachedException.class, () -> sharePartitionManager.newContext("grp", reqData, EMPTY_PART_LIST,
            reqMetadata2, false, "id-2"));
    }

    @Test
    public void testNewContextExistingSessionNewRequestWithInitialEpoch() {
        ShareSessionCache cache = new ShareSessionCache(10);
        sharePartitionManager = SharePartitionManagerBuilder.builder()
            .withCache(cache)
            .build();

        Uuid tpId0 = Uuid.randomUuid();
        TopicIdPartition tp0 = new TopicIdPartition(tpId0, new TopicPartition("foo", 0));
        TopicIdPartition tp1 = new TopicIdPartition(tpId0, new TopicPartition("foo", 1));

        String groupId = "grp";
        Uuid memberId = Uuid.randomUuid();
        List<TopicIdPartition> reqData = List.of(tp0, tp1);

        ShareRequestMetadata reqMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH);