如何在Apache Beam Java中将TestStreams与MultiOutput类一起使用

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-core</artifactId>
    <version>2.33.0</version>
</dependency>
<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-test-streams</artifactId>
    <version>2.33.0</version>
    <scope>test</scope>
</dependency>

import org.apache.beam.sdk.testing.TestStream;
import org.apache.beam.sdk.testing.TestStream.Event;
import org.apache.beam.sdk.testing.TestStream.Builder;
import org.apache.beam.sdk.testing.TestStream.ElementEvent;
import org.apache.beam.sdk.testing.TestStream.Event.Type;
import org.apache.beam.sdk.testing.TestStream.ProcessingTimeEvent;
import org.apache.beam.sdk.testing.TestStream.WatermarkEvent;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TupleTagList;

import static org.apache.beam.sdk.testing.PAssert.that;
import static org.apache.beam.sdk.testing.TestStream.create;

@Test
public void testMultiOutput() {
    // 创建一个TestStream对象
    TestStream<String> testStream = create(StringUtf8Coder.of())
            .addElements("element1", "element2", "element3")
            .advanceWatermarkToInfinity();

    // 创建一个MultiOutput对象
    TupleTag<String> mainOutputTag = new TupleTag<>();
    TupleTag<String> additionalOutputTag = new TupleTag<>();
    MultiOutput<String> multiOutput = MultiOutput.withTags(mainOutputTag, additionalOutputTag);

    // 创建一个PCollection对象，并应用数据处理逻辑
    PCollection<String> output = pipeline.apply(testStream)
            .apply(ParDo.of(new MyDoFn()).withOutputTags(mainOutputTag, TupleTagList.of(additionalOutputTag)));

    // 验证输出结果
    that(output).containsInAnyOrder("element1", "element2", "element3");

    // 获取额外的输出结果
    PCollection<String> additionalOutput = output.get(additionalOutputTag);
    that(additionalOutput).empty();
}

// 自定义的DoFn类，用于处理数据流
public static class MyDoFn extends DoFn<String, String> {
    @ProcessElement
    public void processElement(ProcessContext c) {
        String element = c.element();
        // 处理数据流的逻辑
        c.output(element);
    }
}