@Override
    public TypedResult<List<RowData>> retrieveChanges() {
        synchronized (resultLock) {
            // retrieval thread is alive return a record if available
            // but the program must not have failed
            if (isRetrieving() && executionException.get() == null) {
                if (changeRecordBuffer.isEmpty()) {
                    return TypedResult.empty();
                } else {
                    final List<RowData> change = new ArrayList<>(changeRecordBuffer);
                    changeRecordBuffer.clear();
                    resultLock.notify();
                    return TypedResult.payload(change);
                }
            }
            // retrieval thread is dead but there is still a record to be delivered
            else if (!isRetrieving() && !changeRecordBuffer.isEmpty()) {
                final List<RowData> change = new ArrayList<>(changeRecordBuffer);
                changeRecordBuffer.clear();
                return TypedResult.payload(change);
            }
            // no results can be returned anymore
            else {
                return handleMissingResult();
            }
        }
    }
@Test
    void testRetrieveChanges() throws Exception {
        int totalCount = ChangelogCollectResult.CHANGE_RECORD_BUFFER_SIZE * 2;
        CloseableIterator<RowData> data =
                CloseableIterator.adapterForIterator(
                        IntStream.range(0, totalCount)
                                .mapToObj(i -> (RowData) GenericRowData.of(i))
                                .iterator());
        ChangelogCollectResult changelogResult =
                new ChangelogCollectResult(
                        new StatementResult(
                                ResolvedSchema.of(Column.physical("id", DataTypes.INT())),
                                data,
                                true,
                                ResultKind.SUCCESS_WITH_CONTENT,
                                JobID.generate()));

        int count = 0;
        boolean running = true;
        while (running) {
            final TypedResult<List<RowData>> result = changelogResult.retrieveChanges();
            Thread.sleep(100); // slow the processing down
            switch (result.getType()) {
                case EMPTY:
                    // do nothing
                    break;
                case EOS:
                    running = false;
                    break;
                case PAYLOAD:
                    count += result.getPayload().size();
                    break;
                default:
                    throw new SqlExecutionException("Unknown result type: " + result.getType());
            }
        }
        assertThat(count).isEqualTo(totalCount);
    }