@ParameterizedTest
    @MethodSource("testErrorInInitializeSource")
    public void testErrorInInitialize(RuntimeException expectedException) {
        buildDependencies();
        subscribeAndAssign(topicAPartition0);

        // Create a ShareFetchCollector that fails on ShareCompletedFetch initialization.
        fetchCollector = new ShareFetchCollector<String, String>(logContext,
                metadata,
                subscriptions,
                fetchConfig,
                deserializers) {

            @Override
            protected ShareCompletedFetch initialize(final ShareCompletedFetch completedFetch) {
                throw expectedException;
            }
        };

        // Add the ShareCompletedFetch to the ShareFetchBuffer queue - the number of records doesn't matter
        ShareCompletedFetch completedFetch = completedFetchBuilder
                .recordCount(10)
                .build();
        fetchBuffer.add(completedFetch);

        // At first, the queue is populated
        assertFalse(fetchBuffer.isEmpty());

        // Now run our ill-fated collectFetch.
        assertThrows(expectedException.getClass(), () -> fetchCollector.collect(fetchBuffer));
    }
This tests the {@link ShareFetchCollector} functionality.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareFetchCollectorTest.java