private void maybeMarkUserPartitionsAsReady() {
        if (isAllUserTopicPartitionsInitialized) {
            return;
        }
        maybeFetchStartAndEndOffsets();
        boolean isAllInitialized = true;
        for (final UserTopicIdPartition utp : assignedUserTopicIdPartitions.values()) {
            if (utp.isAssigned && !utp.isInitialized) {
                final Integer metadataPartition = utp.metadataPartition;
                final StartAndEndOffsetHolder holder = offsetHolderByMetadataPartition.get(toRemoteLogPartition(metadataPartition));
                // The offset-holder can be null, when the recent assignment wasn't picked up by the consumer.
                if (holder != null) {
                    final Long readOffset = readOffsetsByMetadataPartition.getOrDefault(metadataPartition, -1L);
                    // 1) The end-offset was fetched only once during reassignment. The metadata-partition can receive
                    // new stream of records, so the consumer can read records more than the last-fetched end-offset.
                    // 2) When the internal topic becomes empty due to breach by size/time/start-offset, then there
                    // are no records to read.
                    if (readOffset + 1 >= holder.endOffset || holder.endOffset.equals(holder.startOffset)) {
                        markInitialized(utp);
                    } else {
                        log.debug("The user-topic-partition {} could not be marked initialized since the read-offset is {} " +
                                "but the end-offset is {} for the metadata-partition {}", utp, readOffset, holder.endOffset,
                            metadataPartition);
                    }
                } else {
                    log.debug("The offset-holder is null for the metadata-partition {}. The consumer may not have picked" +
                            " up the recent assignment", metadataPartition);
                }
            }
            isAllInitialized = isAllInitialized && utp.isAssigned && utp.isInitialized;
        }
        if (isAllInitialized) {
            log.info("Initialized for all the {} assigned user-partitions mapped to the {} meta-partitions in {} ms",
                assignedUserTopicIdPartitions.size(), assignedMetadataPartitions.size(),
                time.milliseconds() - uninitializedAt);
        }
        isAllUserTopicPartitionsInitialized = isAllInitialized;
    }
@Test
    public void testMaybeMarkUserPartitionsAsReady() {
        final TopicIdPartition tpId = getIdPartitions("hello", 1).get(0);
        final int metadataPartition = partitioner.metadataPartition(tpId);
        consumer.updateEndOffsets(Collections.singletonMap(toRemoteLogPartition(metadataPartition), 2L));
        consumerTask.addAssignmentsForPartitions(Collections.singleton(tpId));
        consumerTask.ingestRecords();

        assertTrue(consumerTask.isUserPartitionAssigned(tpId), "Partition " + tpId + " has not been assigned");
        assertTrue(consumerTask.isMetadataPartitionAssigned(metadataPartition));
        assertFalse(handler.isPartitionInitialized.containsKey(tpId));
        IntStream.range(0, 5).forEach(offset -> addRecord(consumer, metadataPartition, tpId, offset));
        consumerTask.ingestRecords();
        assertEquals(Optional.of(4L), consumerTask.readOffsetForMetadataPartition(metadataPartition));
        assertTrue(handler.isPartitionInitialized.get(tpId));
    }