Hi I am trying to join two datastreams from Kafka and output to a sink, however my joined stream does not output any data. In the overview window of the Flink UI, it also looks like my sink doesn't exist and after a few minutes my job crashes and runs out of memory, What is wrong with my join logic?

DataStream<Enhanced> joinedStream = EntriesStream.join(historyStream)
				.where(new KeySelector<GenericRecord, String>() {

					@Override
					public String getKey(GenericRecord value) throws Exception {
						return value.get("la_id").toString();

					}
				}).equalTo(new KeySelector<GenericRecord, String>() {

					@Override
					public String getKey(GenericRecord value) throws Exception {
						return value.get("id").toString();
					}
				}).window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
				.apply(new JoinFunction<GenericRecord, GenericRecord, Enhanced>() {


					@Override
					public Enhanced join(GenericRecord first, GenericRecord second) throws Exception {
						return new Enhanced(
								Long.parseLong(first.get("c_at").toString()),
								first.get("c_type").toString(),
								first.get("id").toString(),
								Integer.parseInt(first.get("d_cts").toString()),
								Integer.parseInt(first.get("c_cts").toString()),
								second.get("prov").toString(),
								second.get("bb_S_T").toString(),
								second.get("p_id").toString(),
								second.get("s_ccurr").toString()
						);
					}
				});

Things I have tried so far: 1. Increased task slots on the task manager 2. Used processing time instead of event time