@Override
    public Map<Integer, ReaderInfo> registeredReaders() {
        final Map<Integer, ReaderInfo> readers = new HashMap<>();
        for (Map.Entry<Integer, ConcurrentMap<Integer, ReaderInfo>> entry :
                registeredReaders.entrySet()) {
            final int subtaskIndex = entry.getKey();
            final Map<Integer, ReaderInfo> attemptReaders = entry.getValue();
            int earliestAttempt = Integer.MAX_VALUE;
            for (int attemptNumber : attemptReaders.keySet()) {
                if (attemptNumber < earliestAttempt) {
                    earliestAttempt = attemptNumber;
                }
            }
            readers.put(subtaskIndex, attemptReaders.get(earliestAttempt));
        }
        return Collections.unmodifiableMap(readers);
    }
@Test
    void testTaskFailureUnregistersReader() throws Exception {
        sourceReady();
        List<ReaderInfo> readerInfo = registerReaders();

        sourceCoordinator.executionAttemptFailed(0, 0, null);
        waitForCoordinatorToProcessActions();

        assertThat(context.registeredReaders())
                .as("Only reader 2 should be registered.")
                .hasSize(2);
        assertThat(context.registeredReaders().get(0)).isNull();
        assertThat(context.registeredReaders().get(1)).isEqualTo(readerInfo.get(1));
        assertThat(context.registeredReaders().get(2)).isEqualTo(readerInfo.get(2));
    }