public class HudiSource implements Source<RowData, HudiSourceSplit, HudiEnumeratorState> {
  
  //...
  @Override
  public Boundedness getBoundedness() {
    return scanContext.isStreaming() ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED;
  }

  @Override
  public SourceReader<T, HudiSourceSplit> createReader(SourceReaderContext readerContext) {
    HudiSourceReaderMetrics metrics =
        new HudiSourceReaderMetrics(readerContext.metricGroup(), tableName);
    return new HudiSourceReader<>(
        emitter, metrics, readerFunction, splitComparator, readerContext);
  }

  @Override
  public SplitEnumerator<HudiSourceSplit, HudiEnumeratorState> createEnumerator(
      SplitEnumeratorContext<HudiSourceSplit> enumContext) {
    return createEnumerator(enumContext, null);
  }

  @Override
  public SplitEnumerator<HudiSourceSplit, HudiEnumeratorState> restoreEnumerator(
      SplitEnumeratorContext<HudiSourceSplit> enumContext, HudiEnumeratorState enumState) {
    return createEnumerator(enumContext, enumState);
  }

  @Override
  public SimpleVersionedSerializer<HudiSourceSplit> getSplitSerializer() {
    return new HudiSourceSplitSerializer(scanContext.caseSensitive());
  }

  @Override
  public SimpleVersionedSerializer<HudiEnumeratorState> getEnumeratorCheckpointSerializer() {
    return new HudiEnumeratorStateSerializer(scanContext.caseSensitive());
  }
  
  // ...
}

class HudiSourceSplitReader implements SplitReader<RecordAndPosition<RowData>, HudiSourceSplit> {
  private final FlinkRowDataReaderContext readerContext;
  private final SerializableComparator<HudiSourceSplit> splitComparator;
  private final int indexOfSubtask;
  private final Queue<HudiSourceSplit> splits;

  private CloseableIterator<RecordsWithSplitIds<RecordAndPosition<RowData>>> currentReader;
  private HudiSourceSplit currentSplit;
  private String currentSplitId;

  HudiSourceSplitReader(
      FlinkRowDataReaderContext readerContext,
      SerializableComparator<HudiSourceSplit> splitComparator,
      SourceReaderContext context) {
    this.readerContext = readerContext;
    this.splitComparator = splitComparator;
    this.indexOfSubtask = context.getIndexOfSubtask();
    this.splits = Queues.newArrayDeque();
  }
  
  @Override
  public RecordsWithSplitIds<RecordAndPosition<T>> fetch() throws IOException {
    metrics.incrementSplitReaderFetchCalls(1);
    if (currentReader == null) {
      HudiSourceSplit nextSplit = splits.poll();
      if (nextSplit != null) {
        currentSplit = nextSplit;
        currentSplitId = nextSplit.splitId();
        currentReader = readerContext.apply(currentSplit);
      } else {
        return new RecordsBySplits<>(Collections.emptyMap(), Collections.emptySet());
      }
    }

    if (currentReader.hasNext()) {
      // Because Iterator#next() doesn't support checked exception,
      // we need to wrap and unwrap the checked IOException with UncheckedIOException
      try {
        return currentReader.next();
      } catch (UncheckedIOException e) {
        throw e.getCause();
      }
    } else {
      return finishSplit();
    }
  }

  @Override
  public void handleSplitsChanges(SplitsChange<HudiSourceSplit> splitsChange) {
    if (!(splitsChange instanceof SplitsAddition)) {
      throw new UnsupportedOperationException(
          String.format("Unsupported split change: %s", splitsChange.getClass()));
    }

    if (splitComparator != null) {
      List<HudiSourceSplit> newSplits = Lists.newArrayList(splitsChange.splits());
      newSplits.sort(splitComparator);
      splits.addAll(newSplits);
    } else {
      splits.addAll(splitsChange.splits());
    }
  }

  @Override
  public void wakeUp() {
  }

  @Override
  public void close() throws Exception {
    currentSplitId = null;
    if (currentReader != null) {
      currentReader.close();
    }
  }

  @Override
  public void pauseOrResumeSplits(
      Collection<HudiSourceSplit> splitsToPause, Collection<HudiSourceSplit> splitsToResume) {
  }
}

public class HudiSourceReader<T>
    extends SingleThreadMultiplexSourceReaderBase<
        RecordAndPosition<T>, T, HudiSourceSplit, HudiSourceSplit> {
  
  // ... 
  @Override
  public void start() {
    // We request a split only if we did not get splits during the checkpoint restore.
    // Otherwise, reader restarts will keep requesting more and more splits.
    if (getNumberOfCurrentlyAssignedSplits() == 0) {
      requestSplit(Collections.emptyList());
    }
  }

  @Override
  protected void onSplitFinished(Map<String, HudiSourceSplit> finishedSplitIds) {
    requestSplit(Lists.newArrayList(finishedSplitIds.keySet()));
  }

  @Override
  protected HudiSourceSplit initializedState(HudiSourceSplit split) {
    return split;
  }

  @Override
  protected HudiSourceSplit toSplitType(String splitId, HudiSourceSplit splitState) {
    return splitState;
  }

  private void requestSplit(Collection<String> finishedSplitIds) {
    context.sendSourceEventToCoordinator(new SplitRequestEvent(finishedSplitIds));
  }
  
  // ...
}

  @Override
  public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
     return new DataStreamScanProviderAdapter() {

         @Override
         public boolean isBounded() {
             return !conf.getBoolean(FlinkOptions.READ_AS_STREAMING);
         }

         @Override
         public DataStream<RowData> produceDataStream(StreamExecutionEnvironment execEnv) {
             @SuppressWarnings("unchecked")
             TypeInformation<RowData> typeInfo =
                     (TypeInformation<RowData>) TypeInfoDataTypeConverter.fromDataTypeToTypeInfo(getProducedDataType());
             OptionsInference.setupSourceTasks(conf, execEnv.getParallelism());
             if (conf.getBoolean(FlinkOptions.READ_AS_STREAMING)) {
                 if (conf.getBoolean(FlinkOptions.USE_SOURCE_V2)) {
                     return HudiSource
                             .properties(properties)
                             .metaClient(metaClient)
                             .schemaManager(internalSchemaManager)
                             .project(getProjectedSchema())
                             .limit(limit)
                             .filters(filters)
                             .flinkConfig(readableConfig)
                             .buildStream(env);
                 } else {
                     // original logic
                 }
             } else {
                 if (conf.getBoolean(FlinkOptions.USE_SOURCE_V2)) {
                     return HudiSource
                             .properties(properties)
                             .metaClient(metaClient)
                             .schemaManager(internalSchemaManager)
                             .project(getProjectedSchema())
                             .limit(limit)
                             .filters(filters)
                             .flinkConfig(readableConfig)
                             .buildStream(env);
                 } else {
                     // original logic
                 }
             }
         }
     };
 }