@Test public void test() { PAssert .that(buildBasePipeline(pipeline) .apply(AutoKVWrapper .of(null, JobImpl.LocalSequenceHolder.cleanAndGet(getClass().getName() + ".test"), "", ""))) .satisfies(values -> { final List<KV<String, Record>> items = StreamSupport .stream(values.spliterator(), false) .sorted(comparing( k -> k.getValue().getArray(Record.class, "b1").iterator().next().getString("foo"))) .collect(toList()); assertEquals(2, items.size()); assertEquals(2, new HashSet<>(items).size()); assertEquals(asList("a", "b"), items .stream() .map(k -> k.getValue().getArray(Record.class, "b1").iterator().next().getString("foo")) .collect(toList())); return null; }); assertEquals(PipelineResult.State.DONE, pipeline.run().waitUntilFinish()); }
public static PTransform<PCollection<Record>, PCollection<KV<String, Record>>> of(final String plugin, final Function<GroupKeyProvider.GroupContext, String> idGenerator, final String component, final String branch) { return new RecordParDoTransformCoderProvider<>(KvCoder.of(StringUtf8Coder.of(), SchemaRegistryCoder.of()), new AutoKVWrapper(idGenerator, component, branch)); }
AutoKVWrapper extends DoFn<Record, KV<String, Record>> { public static PTransform<PCollection<Record>, PCollection<KV<String, Record>>> of(final String plugin, final Function<GroupKeyProvider.GroupContext, String> idGenerator, final String component, final String branch) { return new RecordParDoTransformCoderProvider<>(KvCoder.of(StringUtf8Coder.of(), SchemaRegistryCoder.of()), new AutoKVWrapper(idGenerator, component, branch)); } }
AutoKVWrapper extends DoFn<Record, KV<String, Record>> { public static PTransform<PCollection<Record>, PCollection<KV<String, Record>>> of(final String plugin, final Function<GroupKeyProvider.GroupContext, String> idGenerator, final String component, final String branch) { return new RecordParDoTransformCoderProvider<>(KvCoder.of(StringUtf8Coder.of(), SchemaRegistryCoder.of()), new AutoKVWrapper(idGenerator, component, branch)); } protected  AutoKVWrapper(); }
AutoKVWrapper extends DoFn<Record, KV<String, Record>> { public static PTransform<PCollection<Record>, PCollection<KV<String, Record>>> of(final String plugin, final Function<GroupKeyProvider.GroupContext, String> idGenerator, final String component, final String branch) { return new RecordParDoTransformCoderProvider<>(KvCoder.of(StringUtf8Coder.of(), SchemaRegistryCoder.of()), new AutoKVWrapper(idGenerator, component, branch)); } protected  AutoKVWrapper(); @ProcessElement void onElement(final ProcessContext context); static PTransform<PCollection<Record>, PCollection<KV<String, Record>>> of(final String plugin,
            final Function<GroupKeyProvider.GroupContext, String> idGenerator, final String component,
            final String branch); }
AutoKVWrapper extends DoFn<Record, KV<String, Record>> { public static PTransform<PCollection<Record>, PCollection<KV<String, Record>>> of(final String plugin, final Function<GroupKeyProvider.GroupContext, String> idGenerator, final String component, final String branch) { return new RecordParDoTransformCoderProvider<>(KvCoder.of(StringUtf8Coder.of(), SchemaRegistryCoder.of()), new AutoKVWrapper(idGenerator, component, branch)); } protected  AutoKVWrapper(); @ProcessElement void onElement(final ProcessContext context); static PTransform<PCollection<Record>, PCollection<KV<String, Record>>> of(final String plugin,
            final Function<GroupKeyProvider.GroupContext, String> idGenerator, final String component,
            final String branch);  }