Hi I am trying to join two Kafka Data Streams from and output to another Kafka topic, however my joined stream does not output any data. After some time, my program crashes and runs out of memory, which I think is a result of the join not working. My code doesn't throw any errors, but the joins don't produce any output. My join logic is below, please suggest possible solutions. P.S: Things I have tried so far: 1. Increased task slots on the task manager 2. Added Watermarks and Timestamps to my Kafka sources 3. Used processing time as my time characteristic

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		// Use event time from the source record
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		// Use S3 as the backend store to store checkpoints for recovery
		env.setStateBackend(new HashMapStateBackend());
		// Perform checkpoint every minute
		env.enableCheckpointing(60000);
		env.getCheckpointConfig().setCheckpointStorage(config.getCheckpointPath());

		KafkaSource<GenericRecord> history_source = KafkaSource.<GenericRecord>builder()
				.setBootstrapServers(BOOTSTRAP_SERVERS)
			    .setTopics("history_topic")
				.setStartingOffsets(OffsetsInitializer.earliest())
				.setGroupId("test-001")
				.setProperty("security.protocol", "SASL_SSL")
				.setProperty("sasl.mechanism", "PLAIN")
				.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='" + CONFLUENT_API_KEY + "' password='" + CONFLUENT_SECRET + "';")
				.setProperty("client.dns.lookup","use_all_dns_ips")
				.setValueOnlyDeserializer(ConfluentRegistryAvroDeserializationSchema.forGeneric(historySchema, SCHEMA_REGISTRY_URL, SCHEMA_REGISTRY_CONFIG))
				.build();

		KafkaSource<GenericRecord> entries_source = KafkaSource.<GenericRecord>builder()
				.setBootstrapServers(BOOTSTRAP_SERVERS)
				.setTopics("entries_topic")
				.setStartingOffsets(OffsetsInitializer.earliest())
				.setGroupId("test-001")
				.setProperty("security.protocol", "SASL_SSL")
				.setProperty("sasl.mechanism", "PLAIN")
				.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='" + CONFLUENT_API_KEY + "' password='" + CONFLUENT_SECRET + "';")
				.setProperty("client.dns.lookup","use_all_dns_ips")
				.setValueOnlyDeserializer(ConfluentRegistryAvroDeserializationSchema.forGeneric(EntriesSchema, SCHEMA_REGISTRY_URL, SCHEMA_REGISTRY_CONFIG))
				.build();

		DataStream<GenericRecord> historyStream = env.fromSource(history_source,  WatermarkStrategy.<GenericRecord>forMonotonousTimestamps().withTimestampAssigner((event, timestamp) -> {
			return timestamp;
		}), "History Source");
		DataStream<GenericRecord> journalEntriesStream = env.fromSource(entries_source,  WatermarkStrategy.<GenericRecord>forMonotonousTimestamps().withTimestampAssigner((event, timestamp) -> {
			return timestamp;
		}), "Entries Source");


 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(TumblingEventTimeWindows.of(Time.seconds(30)))
				.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()
						);
					}
				});

		joinedStream.print();