@Test
    public void testConcurrentFetchMetadataWithoutConcurrentModificationException() {
        S3StreamsMetadataImage streamsImage = generateStreamImage(STREAM0,
            Range.closedOpen(0L, 100000L),
            Range.closedOpen(100000L, 110000L), 20);

        long startOffset = streamsImage.getStreamMetadata(STREAM0).getStartOffset();
        long endOffset = streamsImage.getStreamMetadata(STREAM0).lastRange().endOffset();

        AtomicBoolean hasException = new AtomicBoolean(false);

        CountDownLatch doneLatch = new CountDownLatch(4);
        CountDownLatch startLatch = new CountDownLatch(1);

        class Item {
            long start;
            long end;
            int limit;

            public Item(long start, long end, int limit) {
                this.start = start;
                this.end = end;
                this.limit = limit;
            }
        }

        ConcurrentHashMap<Item, InRangeObjects> result = new ConcurrentHashMap<>();
        ExecutorService es = Executors.newFixedThreadPool(4);
        for (int j = 0; j < 4; j++) {
            es.submit(() -> {
                try {
                    startLatch.await();
                } catch (InterruptedException e) {
                    // ignore
                    return;
                }
                ThreadLocalRandom r = ThreadLocalRandom.current();
                for (int i = 0; i < 10000 && !hasException.get(); i++) {
                    try {
                        long start = r.nextLong(startOffset, endOffset);
                        long end = r.nextLong(start, endOffset);
                        int limit = r.nextInt(3, 17);
                        Item item = new Item(start, end, limit);
                        result.put(item, streamsImage.getObjects(STREAM0, start, end, limit, defaultRangeGetter).get());
                    } catch (Exception e) {
                        hasException.set(true);
                    }
                }

                doneLatch.countDown();
            });
        }

        startLatch.countDown();
        try {
            doneLatch.await();
        } catch (InterruptedException e) {
            //
        } finally {
            es.shutdown();
        }

        assertFalse(hasException.get());

        result.entrySet().forEach(entry -> {
            Item item = entry.getKey();
            InRangeObjects objects = entry.getValue();
            try {
                assertEquals(streamsImage.getObjects(STREAM0, item.start, item.end, item.limit, defaultRangeGetter).get(), objects);
            } catch (Throwable e) {
                throw new RuntimeException(e);
            }
        });
    }
Test get objects with the first hit object is a stream set object.
metadata/src/test/java/org/apache/kafka/image/S3StreamsMetadataImageTest.java