Can I use a flink job to split and send data to multiple jobs according to the incoming data via kafka?

I'm new to Flink. So, this is a flink job which consumes data from kafka and split the data according to some keys and values.

This is the SplitJob Class

public class SplitJob {
public static void main(String[] args) throws Exception {   
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
ObjectMapper mapper = new ObjectMapper();
long startTime = System.currentTimeMillis();
//For recieving data from kafka
//        KafkaSource<String> source = KafkaSource.<String>builder().setBootstrapServers("localhost:9092").setTopics("input-topic").setGroupId("my-group").setStartingOffsets(OffsetsInitializer.latest()).setValueOnlyDeserializer(new SimpleStringSchema()).build();
//        DataStream<String> sourceStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
DataStream<String> sourceStream = env.fromElements("{\"deviceId\":\"0\",\"sessionId\":\"0\",\"component\":[DATAS....}]}");
DataStream<ObjectNode> jsonStream = sourceStream.map(value -> mapper.readTree(value).deepCopy());
DataStream<String> data1 = processComponentTwo(jsonStream, "data1", "hr");
DataStream<String> data2 = processComponentTwo(jsonStream, "data2", "hr");
DataStream<String> data3 = processComponentTwo(jsonStream, "data3", "value");
Job1.consumeDataStream(data1, data3);
Job2.consumeDataStream(data2, data3);
long endTime = System.currentTimeMillis();
long elapsedTime = endTime - startTime;
System.out.println("Elapsed time in splitjob:::::: " + elapsedTime + "ms");
env.execute("SplitJob");
}
private static void processComponent(ObjectNode jsonNodes, String code, String field, Collector<String> collector) {
List<String> resultList = new ArrayList<>();
for (JsonNode component : jsonNodes.get("component")) {
String componentCode = component.get("code").asText();
if (componentCode.equals(code)) {
resultList.add(component.get(field).toString());
}
}
collector.collect(resultList.toString());
}
private static DataStream<String> processComponentTwo(DataStream<ObjectNode> jsonStream, String componentCode, String field) {
return jsonStream.flatMap(new FlatMapFunction<ObjectNode, String>() {
@Override
public void flatMap(ObjectNode jsonNodes, Collector<String> collector) throws Exception {
List<String> resultList = new ArrayList<>();
for (JsonNode component : jsonNodes.get("component")) {
String code = component.get("code").asText();
if (code.equals(componentCode)) {
resultList.add(component.get(field).toString());
}
}
collector.collect(resultList.toString());
}
});
}
}
}

> This is the job1 class

public class Job1 {
public static DataStream<String> consumeDataStream(DataStream<String> data, DataStream<String> pdata) throws Exception {
StreamExecutionEnvironment env2 = StreamExecutionEnvironment.getExecutionEnvironment();
env2.setParallelism(2);
ArrayList<String> cdata = new ArrayList<>();
Iterator<String> cOutput = DataStreamUtils.collect(data);
while (cOutput.hasNext())
cdata.add(cOutput.next());
ArrayList<String> pList = new ArrayList<>();
Iterator<String> pOutput = DataStreamUtils.collect(pdata);
while (parentOutput.hasNext())
pList.add(pOutput.next());
HashMap<String, ArrayList<String>> mergedData = new HashMap<>();
mergedData.put("fhr1", cdata);
mergedData.put("utrine", pList);
DataStream<String> streams = env2.fromElements(mergedData.toString());
DataStream<String> processedStream = streams.map(new MapFunction<String, String>() {
@Override
public String map(String value) throws Exception {
RequestSpecification request = RestAssured.given();
request.header("Content-Type", "application/json");
request.body(mergedData.toString());
request.post("http://localhost:5002/fhr1");
return value.concat(" job 1");
}
});
env2.execute("Job 1");
return fhr;
}
}

Similar to this we have job2 class. Not familiar with flink so I don't know if this is the correct way to use it. Anyway what I'm trying to do is, Get data from kafka, split it and send to multiple jobs like job1, and I want to send the data from job1 to a flask api. This sending data to flask API should be on every other jobs created. so this is taking way too much time, the methods used here are scrapped from internet.
Also we need to scale up the jobs like job1 according to the data splitting.

The flink version using is : 1.17.0

I have tried multithreading in splitjob class, I don't think it worked.
Currently using a hardcoded data other that data from kafka.
The overall structure is working but the performance is very poor, and it is taking way too much time to complete. I need to reduce the time to below 10ms, currently its taking 10s.
Is it even possible or should I try any other methods.

Well, there are multiple ways to achieve this which might be really opinionated.

When running multiple jobs (execute() call) under the same application, it actually blocks. So your call to another execute() will wait the previous execute() call to complete. See here.

> Compared to the Per-Job (deprecated) mode, the Application Mode allows the submission of applications consisting of multiple jobs. The order of job execution is not affected by the deployment mode but by the call used to launch the job. Using execute(), which is blocking, establishes an order and it will lead to the execution of the “next” job being postponed until “this” job finishes. Using executeAsync(), which is non-blocking, will lead to the “next” job starting before “this” job finishes.

>The Application Mode allows for multi-execute() applications but High-Availability is not supported in these cases. High-Availability in Application Mode is only supported for single-execute() applications.

>Additionally, when any of multiple running jobs in Application Mode (submitted for example using executeAsync()) gets cancelled, all jobs will be stopped and the JobManager will shut down. Regular job completions (by the sources shutting down) are supported.

Apart of the above, I am not fan of running multiple job inside one application since it does not offer high-availability. So I would normally do either:

1. I am thinking you can use side outputs here. The rough idea would be to put splitted records into different side output. Then from each independent side output, you can do your further processing.

2. In your main app, just split the records and produce back to different kafka topic for each job type. Then you can create another job that consume from that particular topic and do the processing.