static @Nullable String resolveConsumerArn(Read spec, PipelineOptions options) {
    String streamName = Preconditions.checkArgumentNotNull(spec.getStreamName());
    KinesisIOOptions sourceOptions = options.as(KinesisIOOptions.class);
    Map<String, String> streamToArnMapping = sourceOptions.getKinesisIOConsumerArns();

    String consumerArn;
    if (streamToArnMapping.containsKey(streamName)) {
      consumerArn = streamToArnMapping.get(streamName); // can resolve to null too
    } else {
      consumerArn = spec.getConsumerArn();
    }

    return consumerArn;
  }
@Test
  public void testConsumerArnForSpecificStreamNotPassedInPipelineOptions() {
    KinesisIO.Read readSpec = KinesisIO.read().withStreamName("stream-xxx");
    KinesisIOOptions options =
        createIOOptions("--kinesisIOConsumerArns={\"stream-01\": \"arn-01\"}");
    assertThat(KinesisSource.resolveConsumerArn(readSpec, options)).isNull();
  }