KStream KTable join does not output messages.

[chainhead]

I am building a project modeled on this project. The key difference is, I want to output, conditionally, a message using the messages from the joined topics. As opposed to the example project, where an aggregation is performed. (I am struggling to use Serde for JSON messages - a separate battle. So, I have simplified the message structure as follows.)

• t1 (KStream) - a plain text value.
• t2 (KTable) - a plain text value separated by a ;.
• t3 (KStream) - a CSV string.

I am publishing messages using kafkacat with the -k option to set a key e.g. k1. I can confirm that the messages are present in t1 and t2 by running kafkacat with the -C option. The problem I am facing is: I don't see any output in t3. Where do I start looking? What do I look for?

This is my TopologyProducer.java.

@Produces
  public Topology buildTopology() {
    StreamsBuilder builder = new StreamsBuilder();

    ObjectMapperSerde<stream1> stream1 = new ObjectMapperSerde<>(stream1.class);
    ObjectMapperSerde<topic1> topic1 = new ObjectMapperSerde<>(topic1.class);
    ObjectMapperSerde<output1> output1 = new ObjectMapperSerde<>(output1.class);

    GlobalKTable<String, topic1> topic1 = builder.globalTable(
        t2,
        Consumed.with(Serdes.String(), topic1));

    builder.stream(t1,
        Consumed.with(Serdes.String(), stream1))
        .join(t2,
            (paramName, paramValue) -> paramName,
            (paramValue, paramLimits) -> {
              // Add some logic to return conditionally
              return new output1("paramName", 0.0, 0.0, true);
            })
        .to(t3,
            Produced.with(Serdes.String(), output1));
    return builder.build();
  }
}

[quarkus-bot[bot]]

/cc @alesj, @gunnarmorling, @ozangunalp, @rquinio

[mariuszbozek]

Hi,

in my case problem was not equal keys. Are you sure that your keys in t1 and t2 are equal?