public class FakeSourceReader extends SimpleSourceReaderBase<Row> {
  private final BitSailConfiguration readerConfiguration;  private final TypeInfo<?>[] typeInfos;
  private final transient int totalCount;  private final transient RateLimiter fakeGenerateRate;  private final transient AtomicLong counter;
  private final FakeRowGenerator fakeRowGenerator;
  public FakeSourceReader(BitSailConfiguration readerConfiguration, Context context) {    this.readerConfiguration = readerConfiguration;    this.typeInfos = context.getTypeInfos();    this.totalCount = readerConfiguration.get(FakeReaderOptions.TOTAL_COUNT);    this.fakeGenerateRate = RateLimiter.create(readerConfiguration.get(FakeReaderOptions.RATE));    this.counter = new AtomicLong();    this.fakeRowGenerator = new FakeRowGenerator(readerConfiguration, context.getIndexOfSubtask());  }
  @Override  public void pollNext(SourcePipeline<Row> pipeline) throws Exception {    fakeGenerateRate.acquire();    pipeline.output(fakeRowGenerator.fakeOneRecord(typeInfos));  }
  @Override  public boolean hasMoreElements() {    return counter.incrementAndGet() <= totalCount;  }}

public class PrintWriter implements Writer<Row, String, Integer> {  private static final Logger LOG = LoggerFactory.getLogger(PrintWriter.class);
  private final int batchSize;  private final List<String> fieldNames;
  private final List<String> writeBuffer;  private final List<String> commitBuffer;
  private final AtomicInteger printCount;
  public PrintWriter(int batchSize, List<String> fieldNames) {    this(batchSize, fieldNames, 0);  }
  public PrintWriter(int batchSize, List<String> fieldNames, int alreadyPrintCount) {    Preconditions.checkState(batchSize > 0, "batch size must be larger than 0");    this.batchSize = batchSize;    this.fieldNames = fieldNames;    this.writeBuffer = new ArrayList<>(batchSize);    this.commitBuffer = new ArrayList<>(batchSize);    printCount = new AtomicInteger(alreadyPrintCount);  }
  @Override  public void write(Row element) {    String[] fields = new String[element.getFields().length];    for (int i = 0; i < element.getFields().length; ++i) {      fields[i] = String.format("\"%s\":\"%s\"", fieldNames.get(i), element.getField(i).toString());    }
    writeBuffer.add("[" + String.join(",", fields) + "]");    if (writeBuffer.size() == batchSize) {      this.flush(false);    }    printCount.incrementAndGet();  }
  @Override  public void flush(boolean endOfInput) {    commitBuffer.addAll(writeBuffer);    writeBuffer.clear();    if (endOfInput) {      LOG.info("all records are sent to commit buffer.");    }  }
  @Override  public List<String> prepareCommit() {    return commitBuffer;  }
  @Override  public List<Integer> snapshotState(long checkpointId) {    return Collections.singletonList(printCount.get());  }}