kafka-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Matthias J. Sax" <matth...@confluent.io>
Subject Re: Problem with multiple joins in one topology
Date Thu, 08 Dec 2016 08:03:05 GMT
Hi Brian,

Sorry for you headache. We are aware that current join semantics in
Streams are not straight forward.

We did rework those already in trunk and this change will be included in
next release 0.10.2. Please build from trunk and let us know if this
resolves your issue.

For details, see this wiki page explaining current and new join semantics:
https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Join+Semantics

For more details see the KIP:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-77%3A+Improve+Kafka+Streams+Join+Semantics


Long story short: leftJoin only triggers a join computation for the left
input while records of the right input only update the (right) input
KTable but do not compute a join result. Thus, if your right input data
arrives before you left input data it works -- however, if you left
input data arrives first, it will not enrich the stream but join with
"null".


-Matthias

On 12/7/16 9:25 AM, Brian Krahmer wrote:
> Hey guys,
> 
>   I'm having a hell of a time here.  I've worked for days trying to get
> this joining pipeline working.  I thought I had it working last week,
> but my jubilation was premature.  The point was to take data in from
> five different topics and merge them together to obtain one enriched
> event (output to compacted topic).  Can anybody spot what I'm doing
> wrong?  The ordering makes no difference.  For example, I've switched
> the locationInput and the vehicleReservedInput inputs in the leftJoin
> calls below, and I get the same results.  The location part of the
> enrichment works while the vehicleReserved part does not.  I can't even
> think of how to restructure the topology without resorting to building
> my own lower-level topology.
> 
> thanks,
> brian
> 
> 
> KTable<String, VehicleFinderData> fleetInput =
> builder.table(Serdes.String(),
>                 vehicleFinderDataSerde, FLEET_TOPIC,
> VEHICLE_ENRICHER_FLEET_STORE);
> ...
> fleetInput.print("fleetInput");
> locationInput.print("locationInput");
> vehicleReservedInput.print("vehicleReservedInput");
> vehicleReleasedInput.print("vehicleReleasedInput");
> vehicleUsageEndedInput.print("vehicleUsageEndedInput");
> 
> KTable<String, VehicleFinderData> mergeStepOne =
> fleetInput.leftJoin(locationInput, VehicleFinderData::merge);
> mergeStepOne.print("mergeStepOne");
> KTable<String, VehicleFinderData> mergeStepTwo =
> mergeStepOne.leftJoin(vehicleReleasedInput, VehicleFinderData::merge);
> mergeStepTwo.print("mergeStepTwo");
> KTable<String, VehicleFinderData> mergeStepThree =
> mergeStepTwo.leftJoin(vehicleUsageEndedInput, VehicleFinderData::merge);
> mergeStepThree.print("mergeStepThree");
> KTable<String, VehicleFinderData> mergeStepFour =
> mergeStepThree.leftJoin(vehicleReservedInput, VehicleFinderData::merge);
> mergeStepFour.print("mergeStepFour");
> 
> ** Generate a location event **
> 
> [locationInput]: 93838671-e591-4849-ae12-6f30cb9ff7bd , ({snipped json
> value}<-null)
> Deserializing from topic VehicleEnricherFleetStore
> Merge operation called
> [mergeStepOne]: 93838671-e591-4849-ae12-6f30cb9ff7bd , ({snipped json
> value}<-null)
> Merge operation called
> [mergeStepTwo]: 93838671-e591-4849-ae12-6f30cb9ff7bd , ({snipped json
> value}<-null)
> Merge operation called
> [mergeStepThree]: 93838671-e591-4849-ae12-6f30cb9ff7bd , ({snipped json
> value}<-null)
> Merge operation called
> [mergeStepFour]: 93838671-e591-4849-ae12-6f30cb9ff7bd , ({snipped json
> value}<-null)
> 
> ** New event correctly serialized **
> 
> -------------------------------------------------------
> 
> ** Generate a vehicleReserved event **
> 
> [vehicleReservedInput]: 93838671-e591-4849-ae12-6f30cb9ff7bd , ({snipped
> json value}<-null)
> [mergeStepFour]: 93838671-e591-4849-ae12-6f30cb9ff7bd , (null<-null)
> 
> ** NO EVENT **
> 
> 


Mime
View raw message