Commit 0cf27ec5 authored by Noel Alonso's avatar Noel Alonso
Browse files

Convierte a table para evitar errores de sessionId 

Al hacer join con windows time puede que inserte elementos que no debe y
ese es el motivo de los errores de session id
parent 26917e60

vessels-commands/src/main/java/es/redmic/vesselscommands/streams/VesselTrackingEventStreams.java

+6 −6
Original line number Diff line number Diff line 
package es.redmic.vesselscommands.streams;



import org.apache.kafka.streams.kstream.GlobalKTable;

import org.apache.kafka.streams.kstream.JoinWindows;

import org.apache.kafka.streams.kstream.KStream;

import org.apache.kafka.streams.kstream.KTable;



import es.redmic.brokerlib.alert.AlertService;

import es.redmic.brokerlib.avro.common.Event;
@@ -262,11 +262,11 @@ public class VesselTrackingEventStreams extends EventSourcingStreams { 
	private void createTrackingFromRealtimeTrackingVessel(KStream<String, VesselTrackingDTO> realTimeTracking,

			KStream<String, Event> events) {



		realTimeTracking

				.leftJoin(events,

		KTable<String, Event> table = events.groupByKey().reduce((aggValue, newValue) -> newValue);



		realTimeTracking.leftJoin(table,

				(vesselTrackingDTO, vesselTrackingEvent) -> getCreateTrackingFromRealtimeTrackingVessel(

								vesselTrackingDTO, vesselTrackingEvent),

						JoinWindows.of(windowsTime))

						vesselTrackingDTO, vesselTrackingEvent))

				.filter((k, v) -> (v != null)).to(topic);

	}