Hi Everyone, I have a straightforward use case but cannot make it work. I am using BATCH mode. I have 2 DataStreams, assume they are like this:

DataStream<Integer> input = env.fromElements(1, 2, 3);
DataStream<Integer> input2 = env.fromElements(1, 2, 3, 4, 5);

I want to take the intersection of these 2 streams. I tried different things, but they seem to fail with error: https://ideone.com/LdtTmE Any easy way to take intersection?

Set<Integer> st = new HashSet<>();
    DataStream<Integer> input = env.fromElements(1, 2, 3);
    CloseableIterator<Integer> integerCloseableIterator = input.executeAndCollect();
    while(integerCloseableIterator.hasNext()) {
      st.add(integerCloseableIterator.next());
    }
    integerCloseableIterator.close();
    DataStream<Integer> input2 = env.fromElements(1, 2, 3, 4, 5)
      .filter(new FilterFunction<Integer>() {
        @Override
        public boolean filter(Integer value) throws Exception {
          return st.contains(value);
        }
      });
    input2.print();
    env.execute();