+ * It demonstrates how to process records in a Kafka topic with varying schema when + * they are produced using RecordName subject name strategy in Confluent Schema Registry. + */ +public class StreamingJob { + private static final Logger LOGGER = LoggerFactory.getLogger(StreamingJob.class); + // Name of the local JSON resource with the application properties in the same format as they are received from the MSF runtime + private static final String LOCAL_APPLICATION_PROPERTIES_RESOURCE = "flink-consumer-dev.json"; + + // Names of the configuration group containing the application properties + private static final String APPLICATION_CONFIG_GROUP = "FlinkApplicationProperties"; + + public static void main(String[] args) throws Exception { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + + if (isLocal(env)) { + env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new org.apache.flink.configuration.Configuration()); + env.enableCheckpointing(60000); + } + + Configuration config = loadConfiguration(env); + + ensureTopicExist(config); + + KafkaSource source = kafkaSource(config); + DataStream options = env.fromSource(source, WatermarkStrategy.noWatermarks(), "source"); + + DataStream roomTemperature = options + .filter((FilterFunction) option -> option.getValue() instanceof RoomTemperature) + .map((MapFunction) option -> (RoomTemperature) option.getValue()) + .keyBy((KeySelector) RoomTemperature::getSensorId) + .filter((FilterFunction) temperatureSample -> temperatureSample.getTemperature() > 30.0); + + DataStream airQuality = options + .filter((FilterFunction) option -> option.getValue() instanceof AirQuality) + .map((MapFunction) option -> (AirQuality) option.getValue()) + .keyBy((KeySelector) AirQuality::getSensorId) + .filter((FilterFunction) airQuality12 -> airQuality12.getQualityIndex() > 5); + + + roomTemperature.print(); + airQuality.print(); + + env.execute("flink-consumer"); + } + + private static boolean isLocal(StreamExecutionEnvironment env) { + return env instanceof LocalStreamEnvironment; + } + + /** + * Load application configuration from the service runtime or from a local resource, when the environment is local + */ + private static Configuration loadConfiguration(StreamExecutionEnvironment env) throws IOException { + Map propertyMap; + + if (isLocal(env)) { + LOGGER.info("Loading application properties from '{}'", LOCAL_APPLICATION_PROPERTIES_RESOURCE); + propertyMap = KinesisAnalyticsRuntime.getApplicationProperties(StreamingJob.class.getClassLoader().getResource(LOCAL_APPLICATION_PROPERTIES_RESOURCE).getPath()); + } else { + LOGGER.info("Loading application configuration from "); + propertyMap = KinesisAnalyticsRuntime.getApplicationProperties(); + } + + Properties properties = propertyMap.get(APPLICATION_CONFIG_GROUP); + + return new Configuration(properties); + } + + /** + * KafkaSource for any AVRO-generated class (SpecificRecord) Confluent schema registry. + * + * @param config Application configuration + * + * @return a KafkaSource instance + */ + private static KafkaSource kafkaSource(Configuration config) { + OptionDeserializationSchema d = new OptionDeserializationSchema(config.getSchemaRegistryUrl()); + return KafkaSource.builder() + .setBootstrapServers(config.getBootstrapServers()) + .setTopics(config.getTopic()) + .setGroupId(config.getConsumerGroupId()) + .setDeserializer(d) + .build(); + } + + private static void ensureTopicExist(Configuration config) throws Exception { + Properties adminClientConfig = new Properties(); + + adminClientConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); + + try (AdminClient adminClient = AdminClient.create(adminClientConfig)) { + // Create the topic if it does not exist + NewTopic newTopic = new NewTopic(config.getTopic(), 1, (short)1); + adminClient.createTopics(Collections.singleton(newTopic)).all().get(); + System.out.println("Topic '" + config.getTopic() + "' created successfully."); + } catch (ExecutionException e) { + if (e.getCause() instanceof TopicExistsException) { + // This is fine because we want the topic to exist + return; + } + throw e; + } + } +} + diff --git a/java/AvroOneTopicManySubjects/flink-consumer/src/main/resources/avro/air-quality.avdl b/java/AvroOneTopicManySubjects/flink-consumer/src/main/resources/avro/air-quality.avdl new file mode 100644 index 0000000..26889e5 --- /dev/null +++ b/java/AvroOneTopicManySubjects/flink-consumer/src/main/resources/avro/air-quality.avdl @@ -0,0 +1,9 @@ +@namespace("com.amazonaws.services.msf.avro") +protocol In { + record AirQuality { + int sensorId; + string room; + float qualityIndex; + timestamp_ms timestamp; + } +} \ No newline at end of file diff --git a/java/AvroOneTopicManySubjects/flink-consumer/src/main/resources/avro/room-temperature.avdl b/java/AvroOneTopicManySubjects/flink-consumer/src/main/resources/avro/room-temperature.avdl new file mode 100644 index 0000000..b32b660 --- /dev/null +++ b/java/AvroOneTopicManySubjects/flink-consumer/src/main/resources/avro/room-temperature.avdl @@ -0,0 +1,9 @@ +@namespace("com.amazonaws.services.msf.avro") +protocol Out { + record RoomTemperature { + int sensorId; + string room; + float temperature; + timestamp_ms timestamp; + } +} \ No newline at end of file diff --git a/java/AvroOneTopicManySubjects/flink-consumer/src/main/resources/flink-consumer-dev.json b/java/AvroOneTopicManySubjects/flink-consumer/src/main/resources/flink-consumer-dev.json new file mode 100644 index 0000000..f8f0d35 --- /dev/null +++ b/java/AvroOneTopicManySubjects/flink-consumer/src/main/resources/flink-consumer-dev.json @@ -0,0 +1,10 @@ +[ + { + "PropertyGroupId": "FlinkApplicationProperties", + "PropertyMap": { + "bootstrap.servers": "localhost:29092", + "source.topic": "sensor-data", + "source.consumer.group.id": "flink-consumer" + } + } +] diff --git a/java/AvroOneTopicManySubjects/flink-consumer/src/main/resources/log4j2.properties b/java/AvroOneTopicManySubjects/flink-consumer/src/main/resources/log4j2.properties new file mode 100644 index 0000000..c7d36aa --- /dev/null +++ b/java/AvroOneTopicManySubjects/flink-consumer/src/main/resources/log4j2.properties @@ -0,0 +1,7 @@ +appender.console.name = ConsoleAppender +appender.console.type = CONSOLE +appender.console.layout.type = PatternLayout +appender.console.layout.pattern = %d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n + +rootLogger.level = INFO +rootLogger.appenderRef.console.ref = ConsoleAppender \ No newline at end of file diff --git a/java/AvroOneTopicManySubjects/pom.xml b/java/AvroOneTopicManySubjects/pom.xml new file mode 100644 index 0000000..d4a2380 --- /dev/null +++ b/java/AvroOneTopicManySubjects/pom.xml @@ -0,0 +1,16 @@ + + + 4.0.0 + + com.amazonaws + avro-one-topic-many-subjects-sample + 1.0 + pom + + producer-app + flink-consumer + + + diff --git a/java/AvroOneTopicManySubjects/producer-app/pom.xml b/java/AvroOneTopicManySubjects/producer-app/pom.xml new file mode 100644 index 0000000..207adcd --- /dev/null +++ b/java/AvroOneTopicManySubjects/producer-app/pom.xml @@ -0,0 +1,185 @@ + + + 4.0.0 + + com.amazonaws + producer-app + 1.0-SNAPSHOT + + + UTF-8 + ${project.basedir}/target + ${project.name} + 11 + ${target.java.version} + com.amazonaws.services.msf.Sensors + + 1.10.2 + 7.9.0 + 3.9.0 + 1.7.32 + 2.17.2 + 5.9.1 + 1.9.0 + + + + + + confluent + https://packages.confluent.io/maven/ + + + + + + org.apache.kafka + kafka-clients + ${kafka.version} + + + org.apache.avro + avro + ${avro.version} + + + io.confluent + kafka-avro-serializer + ${confluent.version} + + + io.confluent + kafka-schema-registry-client + ${confluent.version} + + + + + org.apache.logging.log4j + log4j-slf4j-impl + ${log4j.version} + + + org.apache.logging.log4j + log4j-api + ${log4j.version} + + + org.apache.logging.log4j + log4j-core + ${log4j.version} + + + + + org.junit.jupiter + junit-jupiter-api + ${junit.version} + test + + + org.junit.jupiter + junit-jupiter-engine + ${junit.version} + test + + + commons-cli + commons-cli + ${commons.version} + + + + + ${buildDirectory} + ${jar.finalName} + + + + org.apache.maven.plugins + maven-compiler-plugin + 3.8.1 + + ${target.java.version} + ${target.java.version} + true + + + + + + org.apache.avro + avro-maven-plugin + ${avro.version} + + + generate-sources + + schema + protocol + idl-protocol + + + + + ${project.basedir}/src/main/resources/avro + ${project.basedir}/src/test/resources/avro + private + String + true + + + + + + + org.apache.maven.plugins + maven-shade-plugin + 3.1.1 + + + + package + + shade + + + + + org.apache.flink:force-shading + com.google.code.findbugs:jsr305 + org.slf4j:* + org.apache.logging.log4j:* + + + + + + *:* + + META-INF/*.SF + META-INF/*.DSA + META-INF/*.RSA + + + + + + + ${main.class} + + + + + + + + + + + \ No newline at end of file diff --git a/java/AvroOneTopicManySubjects/producer-app/src/main/java/com/amazonaws/services/msf/Sensors.java b/java/AvroOneTopicManySubjects/producer-app/src/main/java/com/amazonaws/services/msf/Sensors.java new file mode 100644 index 0000000..5b9d9b5 --- /dev/null +++ b/java/AvroOneTopicManySubjects/producer-app/src/main/java/com/amazonaws/services/msf/Sensors.java @@ -0,0 +1,159 @@ +package com.amazonaws.services.msf; + +import com.amazonaws.services.msf.avro.AirQuality; +import com.amazonaws.services.msf.avro.RoomTemperature; +import io.confluent.kafka.serializers.KafkaAvroSerializer; +import io.confluent.kafka.serializers.subject.RecordNameStrategy; +import org.apache.commons.cli.*; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Instant; +import java.util.*; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Sample application to simulate sensors writing air quality and room temperatures to a Kafka topic. + */ +public class Sensors { + private static final Logger LOG = LoggerFactory.getLogger(Sensors.class); + private static final String DEFAULT_BOOTSTRAP_SERVER = "localhost:29092"; + private static final String DEFAULT_TOPIC_NAME = "sensor-data"; + private static final Long DEFAULT_SLEEP_TIME_BETWEEN_RECORDS = 1000L; + private static final String DEFAULT_SCHEMA_REGISTRY_URL = "http://localhost:8085/"; + + private final String bootstrapServer; + private final String topicName; + private final String schemaRegistry; + private final long sleepTimeBetweenRecordsMillis; + + public static void main(String[] args) throws Exception { + Options options = new Options() + .addOption(Option.builder() + .longOpt("bootstrapServer") + .hasArg() + .desc("Bootstrap server (default: " + DEFAULT_BOOTSTRAP_SERVER + ")") + .build()) + .addOption(Option.builder() + .longOpt("topicName") + .hasArg() + .desc("Topic name (default: " + DEFAULT_TOPIC_NAME + ")") + .build()) + .addOption(Option.builder() + .longOpt("schemaRegistry") + .hasArg() + .desc("Confluent Schema Registry URL (default: " + DEFAULT_SCHEMA_REGISTRY_URL + ")") + .build()) + .addOption(Option.builder() + .longOpt("sleep") + .hasArg() + .desc("Sleep duration in seconds (default: " + DEFAULT_SLEEP_TIME_BETWEEN_RECORDS + ")") + .build()); + + CommandLineParser parser = new DefaultParser(); + HelpFormatter formatter = new HelpFormatter(); + + try { + CommandLine cmd = parser.parse(options, args); + String bootstrapServer = cmd.getOptionValue("bootstrapServer", DEFAULT_BOOTSTRAP_SERVER); + String topicName = cmd.getOptionValue("topicName", DEFAULT_TOPIC_NAME); + String schemaRegistry = cmd.getOptionValue("schemaRegistry", DEFAULT_SCHEMA_REGISTRY_URL); + long sleepTimeBetweenRecordsMillis = Long.parseLong(cmd.getOptionValue("sleep", String.valueOf(DEFAULT_SLEEP_TIME_BETWEEN_RECORDS))); + + LOG.info("BootstrapServer: {}, TopicName: {}", bootstrapServer, topicName); + LOG.info("SchemaRegistry: {}", schemaRegistry); + LOG.info("SleepTimeBetweenRecords: {} ms", sleepTimeBetweenRecordsMillis); + + Sensors sensors = new Sensors(bootstrapServer, topicName, schemaRegistry, sleepTimeBetweenRecordsMillis); + sensors.produce(); + } catch (ParseException e) { + System.err.println("Error parsing command line arguments: " + e.getMessage()); + formatter.printHelp("Sensors", options); + System.exit(1); + } catch (NumberFormatException e) { + System.err.println("Error: sleep parameter must be a valid integer"); + formatter.printHelp("Sensors", options); + System.exit(1); + } + } + + public Sensors(String bootstrapServer, String topicName, String schemaRegistry, long sleepTimeBetweenRecordsMillis) { + if (bootstrapServer == null || bootstrapServer.trim().isEmpty()) { + throw new IllegalArgumentException("bootstrapServer cannot be null or empty"); + } + if (topicName == null || topicName.trim().isEmpty()) { + throw new IllegalArgumentException("topicName cannot be null or empty"); + } + if (schemaRegistry == null || schemaRegistry.trim().isEmpty()) { + throw new IllegalArgumentException("schemaRegistry cannot be null or empty"); + } + if (sleepTimeBetweenRecordsMillis < 0) { + throw new IllegalArgumentException("sleepTimeBetweenRecordsMillis must be non-negative"); + } + + this.bootstrapServer = bootstrapServer; + this.topicName = topicName; + this.schemaRegistry = schemaRegistry; + this.sleepTimeBetweenRecordsMillis = sleepTimeBetweenRecordsMillis; + } + + private void produce() throws Exception { + Properties properties = new Properties(); + properties.put("bootstrap.servers", bootstrapServer); + properties.put("schema.registry.url", schemaRegistry); + properties.put("key.serializer", StringSerializer.class.getName()); + properties.put("value.serializer", KafkaAvroSerializer.class.getName()); + properties.put("value.subject.name.strategy", RecordNameStrategy.class.getName()); + + AtomicBoolean running = new AtomicBoolean(true); + + try (KafkaProducer producer = new KafkaProducer<>(properties)) { + Random random = new Random(); + + while (running.get()) { + // Alternate between AirQuality and RoomTemperature records + boolean isAirQuality = random.nextBoolean(); + + String roomId = "Room-" + random.nextInt(10); + Instant timestamp = Instant.now(); + + // Create record using generated POJOs + Object record; + if (isAirQuality) { + AirQuality airQuality = new AirQuality(); + airQuality.setRoom(roomId); + airQuality.setQualityIndex(400 + random.nextFloat() * 600); // CO2 between 400-1000 ppm + airQuality.setTimestamp(timestamp); + record = airQuality; + } else { + RoomTemperature roomTemp = new RoomTemperature(); + roomTemp.setRoom(roomId); + roomTemp.setTemperature(18 + random.nextFloat() * 22); // Temperature between 18-40°C + roomTemp.setTimestamp(timestamp); + record = roomTemp; + } + + // Create ProducerRecord with the topic name and record + ProducerRecord producerRecord = + new ProducerRecord<>(topicName, record); + + // Send the record asynchronously + producer.send(producerRecord, (metadata, exception) -> { + if (exception != null) { + LOG.error("Error sending record: ", exception); + running.set(false); + } + }); + + Thread.sleep(sleepTimeBetweenRecordsMillis); + } + } catch (InterruptedException e) { + LOG.warn("Producer interrupted", e); + throw e; + } + } + +} diff --git a/java/AvroOneTopicManySubjects/producer-app/src/main/resources/avro/air-quality.avdl b/java/AvroOneTopicManySubjects/producer-app/src/main/resources/avro/air-quality.avdl new file mode 100644 index 0000000..26889e5 --- /dev/null +++ b/java/AvroOneTopicManySubjects/producer-app/src/main/resources/avro/air-quality.avdl @@ -0,0 +1,9 @@ +@namespace("com.amazonaws.services.msf.avro") +protocol In { + record AirQuality { + int sensorId; + string room; + float qualityIndex; + timestamp_ms timestamp; + } +} \ No newline at end of file diff --git a/java/AvroOneTopicManySubjects/producer-app/src/main/resources/avro/room-temperature.avdl b/java/AvroOneTopicManySubjects/producer-app/src/main/resources/avro/room-temperature.avdl new file mode 100644 index 0000000..b32b660 --- /dev/null +++ b/java/AvroOneTopicManySubjects/producer-app/src/main/resources/avro/room-temperature.avdl @@ -0,0 +1,9 @@ +@namespace("com.amazonaws.services.msf.avro") +protocol Out { + record RoomTemperature { + int sensorId; + string room; + float temperature; + timestamp_ms timestamp; + } +} \ No newline at end of file diff --git a/java/AvroOneTopicManySubjects/producer-app/src/main/resources/log4j2.properties b/java/AvroOneTopicManySubjects/producer-app/src/main/resources/log4j2.properties new file mode 100644 index 0000000..c7d36aa --- /dev/null +++ b/java/AvroOneTopicManySubjects/producer-app/src/main/resources/log4j2.properties @@ -0,0 +1,7 @@ +appender.console.name = ConsoleAppender +appender.console.type = CONSOLE +appender.console.layout.type = PatternLayout +appender.console.layout.pattern = %d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n + +rootLogger.level = INFO +rootLogger.appenderRef.console.ref = ConsoleAppender \ No newline at end of file