在Hazelcast jet中，如何将对象映射到BatchSource？

import com.hazelcast.jet.core.BatchSource;
import com.hazelcast.jet.core.ProcessorMetaSupplier;
import com.hazelcast.jet.core.WatermarkSourceUtil;
import com.hazelcast.jet.pipeline.SourceBuilder;
import com.hazelcast.jet.pipeline.StreamSource;

import java.util.List;

public class MyBatchSource<T> implements BatchSource<T> {

    private List<T> data;

    public MyBatchSource(List<T> data) {
        this.data = data;
    }

    @Override
    public void fillBufferFn(SourceBuilder.SourceBuffer<T> sourceBuffer) {
        for (T item : data) {
            sourceBuffer.add(item);
        }
        sourceBuffer.close();
    }

    @Override
    public ProcessorMetaSupplier createSnapshotFn() {
        return WatermarkSourceUtil.NO_SNAPSHOT;
    }

    @Override
    public void restoreSnapshotFn(ProcessorMetaSupplier.Context context) {
        // No-op
    }

    @Override
    public void destroyFn() {
        // No-op
    }

    public static <T> StreamSource<T> source(List<T> data) {
        return SourceBuilder
                .batch("myBatchSource", ctx -> new MyBatchSource<>(data))
                .fillBufferFn(MyBatchSource::fillBufferFn)
                .createSnapshotFn(MyBatchSource::createSnapshotFn)
                .restoreSnapshotFn(MyBatchSource::restoreSnapshotFn)
                .destroyFn(MyBatchSource::destroyFn)
                .build();
    }
}