@Test public void emptyLrTest() { SerializableSupplier<KeyValueStore<LatLng, GeocodeResponse>> geocodeKvStore = () -> GeocodeKvStore.create(new KeyValueTestStoreStub<>()); ExtendedRecord er = ExtendedRecord.newBuilder().setId("777").build(); MetadataRecord mdr = MetadataRecord.newBuilder().setId("777").build(); PCollectionView<MetadataRecord> metadataView = p.apply("Create test metadata", Create.of(mdr)) .apply("Convert into view", View.asSingleton()); PCollection<LocationRecord> recordCollection = p.apply(Create.of(er)) .apply( LocationTransform.builder() .geocodeKvStoreSupplier(geocodeKvStore) .metadataView(metadataView) .create() .interpret()) .apply("Cleaning Date created", ParDo.of(new RemoveDateCreated())); PAssert.that(recordCollection).empty(); p.run(); }
@Override public SingleOutput<ExtendedRecord, LocationRecord> interpret() { return ParDo.of(this).withSideInputs(metadataView); }
LocationTransform extends Transform<ExtendedRecord, LocationRecord> { @Override public SingleOutput<ExtendedRecord, LocationRecord> interpret() { return ParDo.of(this).withSideInputs(metadataView); } }
LocationTransform extends Transform<ExtendedRecord, LocationRecord> { @Override public SingleOutput<ExtendedRecord, LocationRecord> interpret() { return ParDo.of(this).withSideInputs(metadataView); } @Builder(buildMethodName = "create") private  LocationTransform(
      SerializableSupplier<KeyValueStore<LatLng, GeocodeResponse>> geocodeKvStoreSupplier,
      PCollectionView<MetadataRecord> metadataView); }
LocationTransform extends Transform<ExtendedRecord, LocationRecord> { @Override public SingleOutput<ExtendedRecord, LocationRecord> interpret() { return ParDo.of(this).withSideInputs(metadataView); } @Builder(buildMethodName = "create") private  LocationTransform(
      SerializableSupplier<KeyValueStore<LatLng, GeocodeResponse>> geocodeKvStoreSupplier,
      PCollectionView<MetadataRecord> metadataView); MapElements<LocationRecord, KV<String, LocationRecord>> toKv(); LocationTransform counterFn(SerializableConsumer<String> counterFn); @Override SingleOutput<ExtendedRecord, LocationRecord> interpret(); @Setup void setup(); @Teardown void tearDown(); @Override Optional<LocationRecord> convert(ExtendedRecord source); @Override @ProcessElement void processElement(ProcessContext c); Optional<LocationRecord> processElement(ExtendedRecord source, MetadataRecord mdr); }
LocationTransform extends Transform<ExtendedRecord, LocationRecord> { @Override public SingleOutput<ExtendedRecord, LocationRecord> interpret() { return ParDo.of(this).withSideInputs(metadataView); } @Builder(buildMethodName = "create") private  LocationTransform(
      SerializableSupplier<KeyValueStore<LatLng, GeocodeResponse>> geocodeKvStoreSupplier,
      PCollectionView<MetadataRecord> metadataView); MapElements<LocationRecord, KV<String, LocationRecord>> toKv(); LocationTransform counterFn(SerializableConsumer<String> counterFn); @Override SingleOutput<ExtendedRecord, LocationRecord> interpret(); @Setup void setup(); @Teardown void tearDown(); @Override Optional<LocationRecord> convert(ExtendedRecord source); @Override @ProcessElement void processElement(ProcessContext c); Optional<LocationRecord> processElement(ExtendedRecord source, MetadataRecord mdr);  }