Flink CEP: Import Errors and Type Mismatches in Stroke Risk Alarm Application

Question

I'm working on a project that uses Apache Flink's Complex Event Processing (CEP) to create a stroke risk alarm system. However, I'm encountering several issues in my Java code, and I'm seeking help to resolve them.

Code Sample:

Here's my Java code in the StrokeRiskAlarm class:

package hes.cs63.CEPMonitor;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.Pattern;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.KafkaSource;
import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
import org.apache.flink.streaming.connectors.kafka.config.TableKafkaConnectorOptions;

import java.util.List;
import java.util.Map;
import java.util.Properties;

public class StrokeRiskAlert {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "stroke-risk-group");

        // Use KafkaSource instead of FlinkKafkaConsumer
        DataStream<String> patientData = env.fromSource(
                KafkaSource.<String>builder()
                        .setBootstrapServers("localhost:9092")
                        .setGroupId("stroke-risk-group")
                        .setTopic("patient-data-topic")
                        .setProperties(properties)
                        .setDeserializer(new SimpleStringSchema())
                        .setStartingOffsets(StartupMode.EARLIEST)
                        .build(),
                "Kafka Source");

        Pattern<String, ?> highRiskPattern = Pattern.<String>begin("first")
                .where(new SimpleCondition<String>() {
                    @Override
                    public boolean filter(String value) {
                        return getValue(value) > 3;
                    }
                });

        DataStream<Map<String, List<String>>> patternStream = CEP.pattern(
                patientData,
                highRiskPattern);

        DataStream<String> strokeRiskAlerts = patternStream.select(new PatternSelectFunction<String, String>() {
            @Override
            public String select(Map<String, List<String>> pattern) throws Exception {
                String userId = pattern.get("first").get(0).split(",")[0].trim();
                int risk = getTotalRisk(pattern.get("first").get(0), pattern.get("middle").get(0),
                        pattern.get("last").get(0));
                return "Stroke Risk Alert: Patient ID - " + userId + ", Risk Level - " + risk;
            }
        });

        strokeRiskAlerts.print();

        env.execute("Stroke Risk Alert Job");
    }

    private static double getValue(String value) {
        // Parse the input value and extract the relevant data for risk calculation
        String[] parts = value.split(",");
        String type = parts[2].trim();
        double measurementValue = Double.parseDouble(parts[3].trim());

        // Implement your logic to check high stroke risk for different measurements
        if (type.equals("HR")) {
            // HeartMeasurement risk calculation logic
            int risk = 0;
            risk += measurementValue <= 50 ? 1 : 0;
            risk += measurementValue <= 40 ? 1 : 0;
            risk += measurementValue >= 91 ? 1 : 0;
            risk += measurementValue >= 110 ? 1 : 0;
            risk += measurementValue >= 131 ? 1 : 0;
            return risk;
        } else if (type.equals("SBP")) {
            // BloodPressureMeasurement risk calculation logic
            int risk = 0;
            risk += measurementValue <= 110 ? 1 : 0;
            risk += measurementValue <= 100 ? 1 : 0;
            risk += measurementValue <= 90 ? 1 : 0;
            risk += measurementValue >= 220 ? 3 : 0;
            return risk;
        } else if (type.equals("TEMP")) {
            // TempMeasurement risk calculation logic
            int risk = 0;
            risk += measurementValue <= 36 ? 1 : 0;
            risk += measurementValue <= 35 ? 2 : 0;
            risk += measurementValue >= 38.1 ? 1 : 0;
            risk += measurementValue >= 39.1 ? 1 : 0;
            return risk;
        }

        // Default risk calculation
        return 0;
    }

    private static int getTotalRisk(String firstValue, String middleValue, String lastValue) {
        // Calculate the total risk based on the values of the first, middle, and last
        // measurements
        int firstRisk = getValue(firstValue);
        int middleRisk = getValue(middleValue);
        int lastRisk = getValue(lastValue);

        return firstRisk + middleRisk + lastRisk;
    }
}

Problem:

I'm encountering the following errors in my code:

1. "The import org.apache.flink.cep.Pattern cannot be resolved" (Line 5)
2. "The import org.apache.flink.cep.conditions cannot be resolved" (Line 7)
3. "The import org.apache.flink.streaming.connectors.kafka.KafkaSource cannot be resolved" (Line 10)
4. "The import org.apache.flink.streaming.connectors.kafka.config.TableKafkaConnectorOptions cannot be resolved" (Line 12)
5. "KafkaSource cannot be resolved" (Line 28)
6. "Pattern cannot be resolved to a type" (Line 38)
7. "Pattern cannot be resolved" (Line 38)
8. "SimpleCondition cannot be resolved to a type" (Line 39)
9. "The method filter(String) of type new SimpleCondition<String>(){} must override or implement a supertype method" (Line 41)
10. "The method select(new PatternSelectFunction<String,String>(){}) is undefined for the type DataStream<Map<String,List<String>>>" (Line 50)
11. "Type mismatch: cannot convert from double to int" (Lines 106, 107, 108)
12. "Unknown word 'Flink'" (Line 26)

What I've Tried:

I've already added the necessary dependencies to my pom.xml file, including Apache Flink and Kafka dependencies, with the correct versions. However, these errors persist, and I'm not sure how to resolve them.

pom.xml:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.example</groupId>
  <artifactId>flink-kafka-stroke-risk</artifactId>
  <version>1.0-SNAPSHOT</version>

  <properties>
    <java.version>11</java.version>
  </properties>

  <dependencies>
    <!-- Apache Flink Dependencies -->
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-java</artifactId>
      <version>1.16.2</version>
    </dependency>

    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-streaming-java</artifactId>
      <version>1.16.2</version>
      <scope>provided</scope>
    </dependency>

    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-clients</artifactId>
      <version>1.16.2</version>
    </dependency>

    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-cep</artifactId>
      <version>1.16.2</version>
      <scope>provided</scope>
    </dependency>


    <!-- Apache Kafka Dependencies -->
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-clients</artifactId>
      <version>3.5.0</version>
    </dependency>

    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-kafka</artifactId>
      <version>1.16.2</version>
    </dependency>


    <!-- Other Dependencies -->
    <!-- Add other dependencies your project may require here -->

  </dependencies>

  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>3.8.0</version>
        <configuration>
          <source>${java.version}</source>
          <target>${java.version}</target>
        </configuration>
      </plugin>
    </plugins>
  </build>
</project>

I'm looking for guidance on how to resolve these import errors and type mismatches in my Apache Flink CEP-based stroke risk alarm application. Any suggestions or insights into what might be causing these issues would be greatly appreciated. Thank you!