KafkaSourceConsumerFn(
      Class<?> connectorClass,
      SourceRecordMapper<T> fn,
      Integer maxRecords,
      Long milisecondsToRun) {
    this.connectorClass = (Class<? extends SourceConnector>) connectorClass;
    this.fn = fn;
    this.maxRecords = maxRecords;
    this.milisecondsToRun = milisecondsToRun;
  }
@Test
  public void testKafkaSourceConsumerFn() {
    Map<String, String> config =
        ImmutableMap.of(
            "from", "1",
            "to", "10",
            "delay", "0.4",
            "topic", "any");

    Pipeline pipeline = Pipeline.create();

    PCollection<Integer> counts =
        pipeline
            .apply(
                Create.of(Lists.newArrayList(config))
                    .withCoder(MapCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of())))
            .apply(
                ParDo.of(
                    new KafkaSourceConsumerFn<>(
                        CounterSourceConnector.class,
                        sourceRecord ->
                            ((Struct) sourceRecord.value()).getInt64("value").intValue(),
                        10)))
            .setCoder(VarIntCoder.of());

    PAssert.that(counts).containsInAnyOrder(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
    pipeline.run().waitUntilFinish();
  }