static <K, V> ReadFromKafkaDoFn<K, V> create(
      ReadSourceDescriptors<K, V> transform,
      TupleTag<KV<KafkaSourceDescriptor, KafkaRecord<K, V>>> recordTag) {
    if (transform.isBounded()) {
      return new Bounded<>(transform, recordTag);
    } else {
      return new Unbounded<>(transform, recordTag);
    }
  }
@Test
  public void testConstructorWithPollTimeout() {
    ReadSourceDescriptors<String, String> descriptors = makeReadSourceDescriptor(consumer);
    // default poll timeout = 1 scond
    ReadFromKafkaDoFn<String, String> dofnInstance = ReadFromKafkaDoFn.create(descriptors, RECORDS);
    Assert.assertEquals(2L, dofnInstance.consumerPollingTimeout);
    // updated timeout = 5 seconds
    descriptors = descriptors.withConsumerPollingTimeout(5L);
    ReadFromKafkaDoFn<String, String> dofnInstanceNew =
        ReadFromKafkaDoFn.create(descriptors, RECORDS);
    Assert.assertEquals(5L, dofnInstanceNew.consumerPollingTimeout);
  }