@Override
  public PTransformReplacement<PCollection<InputT>, WriteFilesResult<DestinationT>>
      getReplacementTransform(
          AppliedPTransform<
                  PCollection<InputT>,
                  WriteFilesResult<DestinationT>,
                  PTransform<PCollection<InputT>, WriteFilesResult<DestinationT>>>
              transform) {
    try {
      WriteFiles<InputT, DestinationT, ?> replacement =
          WriteFiles.to(WriteFilesTranslation.getSink(transform))
              .withSideInputs(WriteFilesTranslation.getDynamicDestinationSideInputs(transform))
              .withSharding(new LogElementShardsWithDrift<>());
      if (WriteFilesTranslation.isWindowedWrites(transform)) {
        replacement = replacement.withWindowedWrites();
      }
      return PTransformReplacement.of(
          PTransformReplacements.getSingletonMainInput(transform), replacement);
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }
@Test
  public void withNoShardingSpecifiedReturnsNewTransform() {
    ResourceId outputDirectory = LocalResources.fromString("/foo", true /* isDirectory */);

    PTransform<PCollection<Object>, WriteFilesResult<Void>> original =
        WriteFiles.to(
            new FileBasedSink<Object, Void, Object>(
                StaticValueProvider.of(outputDirectory),
                DynamicFileDestinations.constant(new FakeFilenamePolicy())) {
              @Override
              public WriteOperation<Void, Object> createWriteOperation() {
                throw new IllegalArgumentException("Should not be used");
              }
            });
    @SuppressWarnings("unchecked")
    PCollection<Object> objs = (PCollection) p.apply(Create.empty(VoidCoder.of()));

    AppliedPTransform<
            PCollection<Object>,
            WriteFilesResult<Void>,
            PTransform<PCollection<Object>, WriteFilesResult<Void>>>
        originalApplication =
            AppliedPTransform.of(
                "write",
                PValues.expandInput(objs),
                Collections.emptyMap(),
                original,
                ResourceHints.create(),
                p);

    assertThat(
        factory.getReplacementTransform(originalApplication).getTransform(),
        not(equalTo((Object) original)));
  }