pubsub输出以触发批处理管道

我制作了java管道，其中一个管道是从主题订阅者读取消息，如果特别喜欢“开始”字符串，则在同一程序中执行另一个管道，读取csv文件，在数据存储区中查找，加密数据并写入csv输出 .

在这个过程中，从任何条件，我都无法将pubsub管道输出传递给批处理管道的起点 .

=======================这个程序正在终止而没有运行 . 虽然没有错误......

如果我删除检查pubsub管道输出的if条件，即低于1，则数据流显示2个管道，一个用于pubsub，另一个用于文件处理 . 即使没有给pub子主题发送任何消息，文件处理管道也会运行，并且在给另一个时没有任何反应...简而言之，只有在完成pubsub管道之后才能触发批处理管道 . 请帮忙 .

代码片段如下：

PCollection<String> pubsubPipeline = 
p.apply(PubsubIO.readStrings().fromTopic(myTpoic))
            .apply("window",
                    Window.into(SlidingWindows//
                            .of(Duration.standardSeconds(30))//
                            .every(Duration.standardSeconds(30)))) //
            .apply("WordsPerLine", ParDo.of(new DoFn<String, String>() {
                @ProcessElement
                public void processElement(ProcessContext c) throws 
Exception {
                    String s = c.element();
                    final String start = "Start";
                    if (start.equals(s)) {
                        c.output(s.toString());
                    } else {
                        LOG.info("Start not found");
                        return;
 //                     throw new Exception();
                    }
                }
            }));
    String start = "Start";
//      String stsubs = pubsubPipeline.toString();
    if (start.equals(pubSubPipeline))
            {
            LOG.info("Come in if condition");   

    LOG.info("Reading input file");
    PCollection<String> lines = p.apply("Read 
  File",TextIO.read().from(input));
    LOG.info("Lookup in the datastore");
    PCollection<HashMap<String, List<Entity>>> entitySet = 
  lines.apply("Query", ParDo.of(new DoFn<String, HashMap<String, 
  List<Entity>>>() {
        @ProcessElement
  ::
  :
  :
    PCollection<String> output = userSet.apply("Print Entity", 
  ParDo.of(new DoFn<User, String>() {
        @ProcessElement
        public void processElement(ProcessContext c) throws Exception {
            User user = c.element();
            if (user != null && user.getEmail() != null && 
  user.getEmail().equals(user.getEncryptedEmail())) {
                user.setEncryptedEmail(null);
            }
            c.output(user.toString());
        }
    }));


    output.apply(TextIO.write().withHeader("User Id,Email,Encrypted Email").to(outputPrefix).withSuffix(".csv").withoutSharding());
    p.run().waitUntilFinish();

            }   

    else
        LOG.info("comes in else condition");
    return;
  }