@SuppressWarnings("dereference.of.nullable")
  public static PTransform<PCollection<Failure>, PDone> getDlqTransform(String fullConfig) {
    List<String> strings = Splitter.on(":").limit(2).splitToList(fullConfig);
    checkArgument(
        strings.size() == 2, "Invalid config, must start with `identifier:`. %s", fullConfig);
    String key = strings.get(0);
    String config = strings.get(1).trim();
    GenericDlqProvider provider = PROVIDERS.get(key);
    checkArgument(
        provider != null, "Invalid config, no DLQ provider exists with identifier `%s`.", key);
    return provider.newDlqTransform(config);
  }
@Test
  @Category(NeedsRunner.class)
  public void testDlq() {
    StoringDlqProvider.reset();
    Failure failure1 = Failure.newBuilder().setError("a").setPayload("b".getBytes(UTF_8)).build();
    Failure failure2 = Failure.newBuilder().setError("c").setPayload("d".getBytes(UTF_8)).build();
    p.apply(Create.of(failure1, failure2))
        .apply(
            GenericDlq.getDlqTransform(
                StoringDlqProvider.ID + ": " + StoringDlqProvider.CONFIG + " "));
    p.run().waitUntilFinish();
    assertThat(StoringDlqProvider.getFailures(), CoreMatchers.hasItems(failure1, failure2));
  }