LeftJoin processing delay

[okomen-foreflight]

Hey there!

I have this topology:
Topologies:
Sub-topology: 0 for global store (will not generate tasks)
Source: KSTREAM-SOURCE-0000000001 (topics: [fleet-flightId-metadata-positions])
--> KTABLE-SOURCE-0000000002
Processor: KTABLE-SOURCE-0000000002 (stores: [fleet-flightId-metadata-positions-STATE-STORE-0000000000])
--> none
<-- KSTREAM-SOURCE-0000000001
Sub-topology: 4
Source: KSTREAM-SOURCE-0000000026 (topics: [ProcessorAggregatorGroup-repartition])
--> PreviousUpdatesAggregated
Processor: PreviousUpdatesAggregated (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000023])
--> KTABLE-TOSTREAM-0000000027
<-- KSTREAM-SOURCE-0000000026
Processor: KTABLE-TOSTREAM-0000000027 (stores: [])
--> KSTREAM-SINK-0000000028
<-- PreviousUpdatesAggregated
Sink: KSTREAM-SINK-0000000028 (topic: fleet-ready-for-processing)
<-- KTABLE-TOSTREAM-0000000027
Sub-topology: 3
Source: KSTREAM-SOURCE-0000000018 (topics: [TransformationAggregatorGroup-repartition])
--> Transform
Processor: Transform (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000015])
--> KTABLE-TOSTREAM-0000000019
<-- KSTREAM-SOURCE-0000000018
Processor: KTABLE-TOSTREAM-0000000019 (stores: [])
--> TransformedStream
<-- Transform
Processor: TransformedStream (stores: [])
--> Enrich
<-- KTABLE-TOSTREAM-0000000019
Processor: Enrich (stores: [])
--> KSTREAM-FILTER-0000000025
<-- TransformedStream
Processor: KSTREAM-FILTER-0000000025 (stores: [])
--> KSTREAM-SINK-0000000024
<-- Enrich
Sink: KSTREAM-SINK-0000000024 (topic: ProcessorAggregatorGroup-repartition)
<-- KSTREAM-FILTER-0000000025
Sub-topology: 2
Source: KSTREAM-SOURCE-0000000006 (topics: [fleet-tail-positions])
--> KSTREAM-MAPVALUES-0000000007
Processor: KSTREAM-MAPVALUES-0000000007 (stores: [])
--> Filter
<-- KSTREAM-SOURCE-0000000006
Processor: Filter (stores: [])
--> GenerateFlightId
<-- KSTREAM-MAPVALUES-0000000007
Processor: GenerateFlightId (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000010])
--> KTABLE-TOSTREAM-0000000011
<-- Filter
Processor: KTABLE-TOSTREAM-0000000011 (stores: [])
--> GeneratedFlightIdStream
<-- GenerateFlightId
Processor: GeneratedFlightIdStream (stores: [])
--> KSTREAM-FOREACH-0000000013, KSTREAM-FILTER-0000000017
<-- KTABLE-TOSTREAM-0000000011
Processor: KSTREAM-FOREACH-0000000013 (stores: [])
--> none
<-- GeneratedFlightIdStream
Processor: KSTREAM-FILTER-0000000017 (stores: [])
--> KSTREAM-SINK-0000000016
<-- GeneratedFlightIdStream
Sink: KSTREAM-SINK-0000000016 (topic: TransformationAggregatorGroup-repartition)
<-- KSTREAM-FILTER-0000000017
Sub-topology: 1
Source: KSTREAM-SOURCE-0000000003 (topics: [fleet-flightId-updated-metadata])
--> KSTREAM-LEFTJOIN-0000000004
Processor: KSTREAM-LEFTJOIN-0000000004 (stores: [])
--> KSTREAM-SINK-0000000005
<-- KSTREAM-SOURCE-0000000003
Sink: KSTREAM-SINK-0000000005 (topic: fleet-flightId-metadata-positions)
<-- KSTREAM-LEFTJOIN-0000000004

Please only look at Sub-topology-1 and 0

Here you can see that I have a topic called fleet-flightid-updated-metadata which contains plenty of data. But when I start the topology, the Leftjoin is rarely done. Do you know of any delays or similar?

Does it take a while for the topology to start up a left join stream?

[LGouellec]

Hi @okomen-foreflight ,

Can you share your complete topology please ?

Best regards,

[okomen-foreflight]

Hey @LGouellec here is my topology:

var metadataTable = builder.GlobalTable<string, byte[], StringSerDes, ByteArraySerDes>(
    KafkaConstants.FLEET_METADATA_TOPIC,
    InMemory.As<string, byte[]>());

var metadataUpdatesStream = builder.Stream<string, byte[], StringSerDes, ByteArraySerDes>(KafkaConstants.FLEET_METADATA_UPDATES_TOPIC);

metadataUpdatesStream.LeftJoin(metadataTable, (flightId, _) => flightId, (updatedMetadataBytes, existingMetadataBytes) =>
    { 
        var existingMetadata = existingMetadataBytes is null || existingMetadataBytes.Length == 0 ? null : MemoryPackSerializer.Deserialize<InfraFlightMetadata>(existingMetadataBytes);
        
        if (existingMetadata is null)
            return updatedMetadataBytes;
    
        var updatedMetadata = MemoryPackSerializer.Deserialize<InfraFlightMetadata>(updatedMetadataBytes)!;
        
        existingMetadata.UpdatedAt = updatedMetadata.UpdatedAt;
        if (updatedMetadata.ACEMetadata is not null)
            existingMetadata.ACEMetadata = updatedMetadata.ACEMetadata;
    
        if (updatedMetadata.ActualDepartureEstimation is not null)
            existingMetadata.ActualDepartureEstimation = updatedMetadata.ActualDepartureEstimation;
    
        if (existingMetadata.FFFlightPlanMetadata is not null)
            existingMetadata.FFFlightPlanMetadata = updatedMetadata.FFFlightPlanMetadata;

        return MemoryPackSerializer.Serialize(existingMetadata);
    })
    .To(KafkaConstants.FLEET_METADATA_TOPIC);

return metadataTable;

What happens is that the metadataUpdatesStream rarely receives updates even though the topic has plenty of messages in it.

Below is my config:

private StreamConfig CreateStreamConfig()
    {
        return new StreamConfig
        {
            ApplicationId = "fleet-consumer",
            ClientId = "fleet-consumer-client",
            BootstrapServers = _secrets.Value.InternalKafkaBootstrapServers,
            AutoOffsetReset = AutoOffsetReset.Latest,
            DefaultKeySerDes = new StringSerDes(),
            DefaultValueSerDes = new ByteArraySerDes(),
            // NOTE: There can be no more threads then the amount of partitions across all deployed instances
            // i.e. if we have 20 partitions (as we do at the time of writing), and two deployed nodes of service, then each should have 10 threads
            NumStreamThreads = _settings.Value.AmountOfStreamThreads,
            DeserializationExceptionHandler = (context, consumeResult, exception) =>
            {
                _rateLimitedLogger.LogError(exception, "Could not consume from topic. Context: {@Context}, {@ConsumeResult}", context, consumeResult);
                return ExceptionHandlerResponse.CONTINUE;
            },
            InnerExceptionHandler = e =>
            {
                _rateLimitedLogger.LogError(e, "Error during topology execution");
                return ExceptionHandlerResponse.CONTINUE;
            },
            ProductionExceptionHandler = report =>
            {
                _rateLimitedLogger.LogError("Could not produce to topic, {@Report}", report);
                return ExceptionHandlerResponse.CONTINUE;
            },
            StateDir = Path.Combine(Directory.GetCurrentDirectory(), "RocksDbState"),
            Logger = new SerilogLoggerFactory(),
            ReplicationFactor = _settings.Value.ReplicationFactor,
            ExposeLibrdKafkaStats = true,
            StatisticsIntervalMs = (int)LibrdStatsRecordInterval.TotalMilliseconds,
        };
    }

[LGouellec]

@okomen-foreflight,

The GlobalTable is updated by another thread, not the same stream thread dedicated to the Stream(..), Table(..), Aggregate(..) etc ... as you can see here.

So the LeftJoin(..) and the GlobalTable updated is not execute on the same thread, when the LeftJoin is triggered if the global store doesn't contains the specific key, the downstream record contains NULL as the left value.

Can you enable DEBUG logging to understand why the flow of processing is too low ?

config.Logger = LoggerFactory.Create(b =>
                {
                    b.SetMinimumLevel(LogLevel.Debug);
                    b.AddConsole();
                })