public @Nullable SubscriptionPath getSubscription() {
    return subscription == null ? null : subscription.get();
  }
@Test
  public void noSubscriptionNoSplitGeneratesSubscription() throws Exception {
    TopicPath topicPath = PubsubClient.topicPathFromName("my_project", "my_topic");
    factory = PubsubTestClient.createFactoryForCreateSubscription();
    PubsubUnboundedSource source =
        new PubsubUnboundedSource(
            factory,
            StaticValueProvider.of(PubsubClient.projectPathFromId("my_project")),
            StaticValueProvider.of(topicPath),
            null /* subscription */,
            null /* timestampLabel */,
            null /* idLabel */,
            false /* needsAttributes */);
    assertThat(source.getSubscription(), nullValue());

    assertThat(source.getSubscription(), nullValue());

    PipelineOptions options = PipelineOptionsFactory.create();
    PubsubSource actualSource = new PubsubSource(source);
    PubsubReader reader = actualSource.createReader(options, null);
    SubscriptionPath createdSubscription = reader.subscription;
    assertThat(createdSubscription, not(nullValue()));

    PubsubCheckpoint checkpoint = reader.getCheckpointMark();
    assertThat(checkpoint.subscriptionPath, equalTo(createdSubscription.getPath()));

    checkpoint.finalizeCheckpoint();
    PubsubCheckpoint deserCheckpoint =
        CoderUtils.clone(actualSource.getCheckpointMarkCoder(), checkpoint);
    assertThat(checkpoint.subscriptionPath, not(nullValue()));
    assertThat(checkpoint.subscriptionPath, equalTo(deserCheckpoint.subscriptionPath));

    PubsubReader readerFromOriginal = actualSource.createReader(options, checkpoint);
    PubsubReader readerFromDeser = actualSource.createReader(options, deserCheckpoint);

    assertThat(readerFromOriginal.subscription, equalTo(createdSubscription));
    assertThat(readerFromDeser.subscription, equalTo(createdSubscription));
  }