@Test public void test() { PAssert.that(buildBasePipeline(pipeline).apply(RecordBranchFilter.of(null, "b1"))).satisfies(values -> { final List<Record> items = StreamSupport.stream(values.spliterator(), false).collect(toList()); assertEquals(2, items.size()); items.forEach(item -> { assertNull(item.get(Object.class, "b2")); final Object b1 = item.get(Object.class, "b1"); assertNotNull(b1); final Collection<Record> records = Collection.class.cast(b1); assertNotNull(records.stream().map(r -> r.getString("foo")).findFirst().get()); }); return null; }); assertEquals(PipelineResult.State.DONE, pipeline.run().waitUntilFinish()); }
public static PTransform<PCollection<Record>, PCollection<Record>> of(final String plugin, final String branchSelector) { final RecordBuilderFactory lookup = ServiceLookup.lookup(ComponentManager.instance(), plugin, RecordBuilderFactory.class); return new RecordParDoTransformCoderProvider<>(SchemaRegistryCoder.of(), new RecordBranchFilter(lookup, branchSelector)); }
RecordBranchFilter extends DoFn<Record, Record> { public static PTransform<PCollection<Record>, PCollection<Record>> of(final String plugin, final String branchSelector) { final RecordBuilderFactory lookup = ServiceLookup.lookup(ComponentManager.instance(), plugin, RecordBuilderFactory.class); return new RecordParDoTransformCoderProvider<>(SchemaRegistryCoder.of(), new RecordBranchFilter(lookup, branchSelector)); } }
RecordBranchFilter extends DoFn<Record, Record> { public static PTransform<PCollection<Record>, PCollection<Record>> of(final String plugin, final String branchSelector) { final RecordBuilderFactory lookup = ServiceLookup.lookup(ComponentManager.instance(), plugin, RecordBuilderFactory.class); return new RecordParDoTransformCoderProvider<>(SchemaRegistryCoder.of(), new RecordBranchFilter(lookup, branchSelector)); } RecordBranchFilter(final RecordBuilderFactory factory, final String branch); protected  RecordBranchFilter(); }
RecordBranchFilter extends DoFn<Record, Record> { public static PTransform<PCollection<Record>, PCollection<Record>> of(final String plugin, final String branchSelector) { final RecordBuilderFactory lookup = ServiceLookup.lookup(ComponentManager.instance(), plugin, RecordBuilderFactory.class); return new RecordParDoTransformCoderProvider<>(SchemaRegistryCoder.of(), new RecordBranchFilter(lookup, branchSelector)); } RecordBranchFilter(final RecordBuilderFactory factory, final String branch); protected  RecordBranchFilter(); @ProcessElement void onElement(final ProcessContext context); static PTransform<PCollection<Record>, PCollection<Record>> of(final String plugin,
            final String branchSelector); }
RecordBranchFilter extends DoFn<Record, Record> { public static PTransform<PCollection<Record>, PCollection<Record>> of(final String plugin, final String branchSelector) { final RecordBuilderFactory lookup = ServiceLookup.lookup(ComponentManager.instance(), plugin, RecordBuilderFactory.class); return new RecordParDoTransformCoderProvider<>(SchemaRegistryCoder.of(), new RecordBranchFilter(lookup, branchSelector)); } RecordBranchFilter(final RecordBuilderFactory factory, final String branch); protected  RecordBranchFilter(); @ProcessElement void onElement(final ProcessContext context); static PTransform<PCollection<Record>, PCollection<Record>> of(final String plugin,
            final String branchSelector);  }