How can I perform a data match between two Kafka topics without any common key?

I have a unique issue at hand. I have to match two different sets of messages in two separate Kafka topics. Based on the match, I have to produce that data to a third topic. Another constraint that I am facing is that the Kafka environment I have to work in uses Kafka 2.3.

• config_topic::configId, config
• event_topic::eventId, event
• user_data_topic::customId, alert

config and event have some matching fields like corpId, secId, fieldId. But, config may or may not have both corpId and secId. At least one is mandatory.
Below is the solution that I have come up with. Please provide your feedbacks and/or better approach.

The only concern with the below approach that I have is data visibility based on different partitions, causing a missing join condition between my KTables and Kstreams. Once, I am sure of the approach I will add the logic to filter out null config data from the rekeyed tables.
I have simplified the below code by removing things like custom serdes, joined, materialized.

//Create KStream on config topic
KStream<Long, Configuration> configStream = builder.stream(CONFIG_TOPIC);
        
//Create KTable with the first key
KTable<Long, List<Configuration>> confTabKeyedWithCaID = configStream.filter((k, v)->v.getCorpId()!=null).
                        groupBy( (k,v) -> v.getCorpId(), Grouped.with() ).
                        aggregate(ArrayList<Configuration>::new, // initializer
                                (aggKey, newValue, aggValue) -> {
                                    aggValue.add(newValue);
                                    return aggValue;
                                }
                                , Materialized.with());
    
//Create KTable with the second key
KTable<String, List<Configuration>> confTabKeyedWithSecID = configStream.filter((k, v)->v.getCorpId()==null &&
                        v.getSecId()!=null).
                        groupBy( (k,v) -> v.getSecId(), Grouped.with() ).
                        aggregate(ArrayList<AMConfiguration>::new, // initializer
                                (aggKey, newValue, aggValue) -> {
                                    aggValue.add(newValue);
                                    return aggValue;
                                }
                                , Materialized.with());
    
//Create KStream from second topic
KStream<Long, Event> eventStream = builder.stream(EVENT_TOPIC, Consumed.with()).
                        filter((key, value) -> workflowFilter(value));
    
//Rekey KStream which we got from 2nd topic with first key
KStream<Long, Event>  caEventStrKeyedWithCaID = eventStream.map((k, v) -> KeyValue.pair(v.getCorpId(), v));
    
//Rekey KStream which we got from 2nd topic with second key
KStream<String, Event>  caEventStrKeyedWithSecID = caEventStream.map((k, v) -> KeyValue.pair(v.getSecId(), v));
        
//Join Stream and Table keyed with 1st key
KStream<Long, List<Enriched>> enAudListKeyedWithCaID = caAudStrKeyedWithCaID.join(alertConfTabKeyedWithCaID,
                        auditListEnricher, joined1);
        
//Join Stream and Table keyed with 2nd key
KStream<String, List<Enriched>> enEventListKeyedWithSecID = caEventStrKeyedWithSecID.join(confTabKeyedWithSecID,
                        eventListEnricher, joined2);
    
//Perform flatMap operation on 1st keyed Stream
KStream<String, Enriched> eventReKeyed1 = eventListKeyedWithCaID.flatMap((key, recordList)->
                        recordList.stream().map( rec -> KeyValue.pair(""+rec.getEmailId()+rec.getCorpId()+rec.getSecId()+rec.getFieldId(), rec)
                        ).collect(Collectors.toSet()));
    
//Perform flatMap operation on 2nd keyed Stream
KStream<String, Enriched> eventReKeyed2 = eventListKeyedWithSecID.flatMap((key, recordList)->
                        recordList.stream().map( rec -> KeyValue.pair(""+rec.getEmailId()+rec.getCorpId()+rec.getSecId()+rec.getFieldId(), rec)
                        ).collect(Collectors.toSet()));
    
//Merge both the streams using outer join   
KStream<String, Enriched> enEventReKeyedStream = (eventReKeyed1.outerJoin(eventReKeyed2, enrichedRecMerger, JoinWindows.of(joinWindowSizeMsDuration), joined3));
        
//Rekey the stream for joining with the alert stream
KTable<String, Enriched> eventReKeyedTable = eventReKeyedStream.filter((k,v)-> v!=null && v.getCorpId()!=null).groupBy( (k, v) -> v.getEmailId()+v.getCorpId()+v.getSecId()+v.getFieldId(),
                                Grouped.with() ).aggregate(Enriched::new, // initializer
                                (aggKey, newValue, aggValue) -> new Enriched(newValue),
                                Materialized.with()).filter((k,v) -> v!=null);
    
//Get a KTable from alert topic
//key: v.getEmailId() + v.getCorpId() + v.getSecId() + v.getFieldId()
KTable<String, Alert> alertKTable = builder.table(ALERT_TOPIC, Consumed.with()).
                        filter((key, value) -> value!=null);
     
//Perform join between enriched and alert data to perform business logic based on update and insert
KTable<String, Alert> eligibleAlertTable = eventReKeyedTable.leftJoin(alertKTable, alertEnricher,
                        Materialized.with()).filter((k,v) -> v!=null);
eligibleAlertTable.toStream().filter((k,v) -> v!=null).to(ALERT_TOPIC, Produced.with());