Hi Everyone, I have a Flink job that reads data from CSV file that will be stored in S3. However, for testing purpose I have local copies of the source data. The same program if switched from RuntimeExecutionMode.STREAMING - to -> RuntimeExecutionMode.BATCH and commenting out line //.monitorContinuously(Duration._ofSeconds_(1)) in AccountReader will work fine. Any thoughts on what am doing wrong, and what am missing to get it running in streaming mode? The program looks like -

public class JoinJob {
// create environments of both APIs
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
env.setParallelism(4);
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

tableEnv.createTemporaryView("user",TargetDataReader.convertSourceToTable(FILE_PATH_USER, env, tableEnv));
tableEnv.createTemporaryView("account",AccountReader.convertSourceToTable(FILE_PATH_ACCOUNT, env, tableEnv));

Table resultTable = tableEnv
  .sqlQuery("select u.id, u.name, u.email, "
    + a.account_name as account_name, a.type as account_type"
    + from user u left join account a on u.id = a.user_id");

DataStream<Row> resultStream = tableEnv.toDataStream(resultTable);
resultStream.print();

env.execute("Executing flink test job");
}

- - code to read account information from CSV (also there is a similar class to read user information) - - 
public class AccountReader {

 public static Table convertSourceToTable(String path, StreamExecutionEnvironment env,
   StreamTableEnvironment tableEnv) {

  FileSource<String> accountSrc = FileSource.forRecordStreamFormat(
      new TextLineInputFormat("UTF-8"),
      new Path(path))
    //monitors folder to read new files. Every second it checks if folder has more files to read
    .monitorContinuously(Duration.ofSeconds(1))
    .setSplitAssigner(FileSource.DEFAULT_SPLIT_ASSIGNER)

    //This is to decide if we need to tell flink to split file and read in chunks. archived files are not splittable.
    .setFileEnumerator((Provider) () -> new NonSplittingRecursiveEnumerator())
    .build();

  DataStream< Account2> accountStream = env.fromSource(accountSrc,
      WatermarkStrategy.forMonotonousTimestamps(),
      "Account as source")
    .flatMap(new FlatMapFunction<String, Account2>() {
     @Override
     public void flatMap(String value, Collector<Account2> out) throws Exception {
      String[] splittedString = value.split(",");
      Account2 account = new Account2();
      account.setType(Integer.parseInt(splittedString[0]));
      account.setName(splittedString[1]);
			 account.setUserId(splittedString[2]);
      out.collect(account);
     }
    });

  return tableEnv.fromDataStream(accountStream).as("type", "account_name");
 }
}

- - error - -

Exception in thread "main" org.apache.flink.table.api.TableException: Table sink '*anonymous_datastream_sink$4*' doesn't support consuming update and delete changes which is produced by node Join(joinType=[LeftOuterJoin], where=[(id = user_id)], select=[id, name, email], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
	at org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram$SatisfyModifyKindSetTraitVisitor.createNewNode(FlinkChangelogModeInferenceProgram.scala:406)

The result of your business logic is a change log stream and your sink doesn’t support that, which is why it throws an error in streaming mode. When you put it in batch, it is an append log and your sink does support that

what would be an alternative sink

Upsert Kafka can deal with change log streams

does that mean an "join" clause constitutes a change log stream

For a regular join, yes. That’s also explained in https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/queries/joins/#regular-joins

Hi @Martijn Visser, why is a changelog stream generated when Kafka is a source? Kafka is an append-only commit log, so I’m wondering why the event stream in Flink is also not append-only?

It's because the result of the query is a changelog. There's a good talk by Timo Walther on this, see (it goes straight to the part where he talks about changelogs, but I can recommend to watch it completely)

Thanks for the link, that was very helpful!