Real-time data streaming using Apache Flink and Kafka

mvn archetype:generate                \
  -DarchetypeGroupId=org.apache.flink   \
  -DarchetypeArtifactId=flink-quickstart-java \
  -DarchetypeVersion=1.18.1

<properties>
	<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
	<flink.version>1.18.1</flink.version>
	<target.java.version>17</target.java.version>
	<maven.compiler.source>${target.java.version}</maven.compiler.source>
	<maven.compiler.target>${target.java.version}</maven.compiler.target>
	<maven-compiler.version>3.1</maven-compiler.version>
	<lombok.version>1.18.32</lombok.version>
</properties>

<dependencies>
		<!-- Apache Flink dependencies -->
		<!-- These dependencies are provided, because they should not be packaged into the JAR file. -->
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-java</artifactId>
			<version>${flink.version}</version>
			<scope>provided</scope>
		</dependency>
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-core</artifactId>
			<version>${flink.version}</version>
			<scope>provided</scope>
		</dependency>
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-connector-base</artifactId>
			<version>${flink.version}</version>
			<scope>provided</scope>
		</dependency>
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-streaming-java</artifactId>
			<version>${flink.version}</version>
			<scope>provided</scope>
		</dependency>
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-clients</artifactId>
			<version>${flink.version}</version>
			<scope>provided</scope>
		</dependency>
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-shaded-guava</artifactId>
			<version>31.1-jre-17.0</version>
		</dependency>

		<!-- Add connector dependencies here. They must be in the default scope (compile). -->
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-connector-kafka</artifactId>
			<version>3.1.0-1.18</version>
		</dependency>
		<dependency>
			<groupId>org.apache.kafka</groupId>
			<artifactId>kafka-clients</artifactId>
			<version>3.4.0</version>
		</dependency>

		<!-- Add logging framework, to produce console output when running in the IDE. -->
		<!-- These dependencies are excluded from the application JAR by default. -->
		<dependency>
			<groupId>org.projectlombok</groupId>
			<artifactId>lombok</artifactId>
			<version>${lombok.version}</version>
			<scope>provided</scope>
		</dependency>

		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-api</artifactId>
			<version>2.0.12</version>
		</dependency>
		<dependency>
			<groupId>ch.qos.logback</groupId>
			<artifactId>logback-classic</artifactId>
			<version>1.5.3</version>
			<scope>runtime</scope>
		</dependency>
	</dependencies>

<?xml version="1.0" encoding="UTF-8"?>
<configuration>

    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <layout class="ch.qos.logback.classic.PatternLayout">
            <Pattern>
                %d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n
            </Pattern>
        </layout>
    </appender>

	<!-- Change this to your package name -->
    <logger name="com.abdelrani" level="debug" additivity="false"> 
        <appender-ref ref="CONSOLE"/>
    </logger>

    <root level="error">
        <appender-ref ref="CONSOLE"/>
    </root>

</configuration>

public class DataStreamJob {

	public static void main(String[] args) throws Exception {
		// Sets up the execution environment, which is the main entry point to building Flink applications.
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		env.fromElements("Hello, World!").print();

		// Execute program, beginning computation.
		env.execute("Starting the stock price updates job ...");
	}
}

Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/flink/streaming/api/environment/StreamExecutionEnvironment
	at com.abdelrani.stockstreams.DataStreamJob.main(DataStreamJob.java:28)

.......... some logs here ..........

5> Hello World!

@Data
@Builder
public class KafkaProperties {
    private String bootstrapServers;
    private String topic;
    private String groupId;
    private String clientId;
    private String autoOffsetReset;
}

public class PropertiesLoader {
    private static final Map<String, String> ENVS = System.getenv();

    public KafkaProperties priceKafkaProperties() {
        return KafkaProperties.builder()
                .bootstrapServers(ENVS.get("PRICE_KAFKA_BOOTSTRAP_SERVERS"))
                .topic(ENVS.get("PRICE_KAFKA_TOPIC"))
                .groupId(ENVS.get("PRICE_KAFKA_GROUP_ID"))
                .clientId(ENVS.get("PRICE_KAFKA_CLIENT_ID"))
                .autoOffsetReset(ENVS.get("PRICE_KAFKA_AUTO_OFFSET_RESET"))
                .build();
    }

    public KafkaProperties stockKafkaProperties() {
        return KafkaProperties.builder()
                .bootstrapServers(ENVS.get("STOCK_KAFKA_BOOTSTRAP_SERVERS"))
                .topic(ENVS.get("STOCK_KAFKA_TOPIC"))
                .groupId(ENVS.get("STOCK_KAFKA_GROUP_ID"))
                .clientId(ENVS.get("STOCK_KAFKA_CLIENT_ID"))
                .autoOffsetReset(ENVS.get("STOCK_KAFKA_AUTO_OFFSET_RESET"))
                .build();
    }

    public KafkaProperties stockUpdateKafkaProperties() {
        return KafkaProperties.builder()
                .bootstrapServers(ENVS.get("STOCK_UPDATE_KAFKA_BOOTSTRAP_SERVERS"))
                .topic(ENVS.get("STOCK_UPDATE_KAFKA_TOPIC"))
                .autoOffsetReset(ENVS.get("STOCK_UPDATE_KAFKA_AUTO_OFFSET_RESET"))
                .build();
    }
}

# Price config
PRICE_KAFKA_BOOTSTRAP_SERVERS=localhost:9092
PRICE_KAFKA_TOPIC=price
PRICE_KAFKA_GROUP_ID=price_group
RICE_KAFKA_CLIENT_ID=price_client
PRICE_KAFKA_AUTO_OFFSET_RESET=latest

#Stock config
STOCK_KAFKA_BOOTSTRAP_SERVERS=localhost:9092
STOCK_KAFKA_TOPIC=stock
STOCK_KAFKA_GROUP_ID=stock_group
STOCK_KAFKA_CLIENT_ID=stock_client
STOCK_KAFKA_AUTO_OFFSET_RESET=latest

# Stock price updates config
STOCK_UPDATE_KAFKA_BOOTSTRAP_SERVERS=localhost:9092
STOCK_UPDATE_KAFKA_TOPIC=stock_update
STOCK_UPDATE_KAFKA_AUTO_OFFSET_RESET=latest

public class SourceBuilder {
    
    public static  KafkaSource<Price> priceSource(KafkaProperties kafkaProperties) {
        return KafkaSource.<Price>builder()
                .setBootstrapServers(kafkaProperties.getBootstrapServers())
                .setTopics(kafkaProperties.getTopic())
                .setValueOnlyDeserializer(new PriceDeserializer())
                .setGroupId(kafkaProperties.getGroupId())
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
                .build();
    }

    public static  KafkaSource<Stock> stockSource(KafkaProperties kafkaProperties) {
        return KafkaSource.<Stock>builder()
                .setBootstrapServers(kafkaProperties.getBootstrapServers())
                .setTopics(kafkaProperties.getTopic())
                .setDeserializer(new StockDeserializer())
                .setGroupId(kafkaProperties.getGroupId())
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
                .setProperty("partition.discovery.interval.ms", "10000") // discover new partitions per 10 seconds
                .build();
    }

    public static KafkaSink<StockUpdate> stockUpdateSink(KafkaProperties kafkaProperties) {
        return KafkaSink.<StockUpdate>builder()
                .setBootstrapServers(kafkaProperties.getBootstrapServers())
                .setRecordSerializer(new StockUpdateSerializer(kafkaProperties.getTopic()))
                .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE) // at least once delivery
                .build();
    }
}

@Slf4j
public class StockDeserializer implements KafkaRecordDeserializationSchema<Stock> {

    @Override
    public void deserialize(ConsumerRecord<byte[], byte[]> record, Collector<Stock> out) throws IOException {
        Stock message = MAPPER.readValue(record.value(), Stock.class);
        if (log.isDebugEnabled()) {
           log.debug("Received stock with symbol: {}", message.getSymbol());
        }
       out.collect(message);
    }

    @Override
    public TypeInformation<Stock> getProducedType() {
        return TypeInformation.of(Stock.class);
    }
}

@Slf4j
public class PriceDeserializer extends AbstractDeserializationSchema<Price>  {

    @Override
    public Price deserialize(byte[] message) throws IOException {
        Price priceMessage = MAPPER.readValue(message, Price.class);
        if (log.isDebugEnabled()) {
            log.debug("Received price message for stock: {}", priceMessage.getSymbol());
        }
        return priceMessage;
    }

}

public class StockUpdateSerializer implements KafkaRecordSerializationSchema<StockUpdate> {

    private final String topic;

    public StockUpdateSerializer(String topic) {
        this.topic = topic;
    }

    @SneakyThrows
    @Nullable
    @Override
    public ProducerRecord<byte[], byte[]> serialize(StockUpdate element, KafkaSinkContext context, Long timestamp) {
        byte[] key = element.getSymbol().getBytes();
        byte[] result = MAPPER.writeValueAsBytes(element);
        return new ProducerRecord<>(topic, key, result);
    }
}

public class JacksonConfiguration {

    public static final ObjectMapper MAPPER = defaultObjectMapper();


    private static ObjectMapper defaultObjectMapper() {
        return JsonMapper.builder()
                .addModules(new JavaTimeModule(), new Jdk8Module())
                .serializationInclusion(NON_NULL)
                .serializationInclusion(NON_EMPTY)
                .propertyNamingStrategy(new PropertyNamingStrategies.SnakeCaseStrategy())
                .enable(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS)
                .enable(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY)
                .enable(DeserializationFeature.UNWRAP_SINGLE_VALUE_ARRAYS)
                .disable(SerializationFeature.FAIL_ON_EMPTY_BEANS)
                .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
                .disable(DeserializationFeature.ADJUST_DATES_TO_CONTEXT_TIME_ZONE)
                .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS).build();
    }
}

public class StockPriceJoiner extends RichCoFlatMapFunction<Stock, Price, StockUpdate> {

    /**
     * The state that is maintained by this process function
     */
    private ValueState<Price> priceState;

    @Override
    public void open(Configuration parameters) throws Exception {
        priceState = getRuntimeContext().getState(new ValueStateDescriptor<>("price", Price.class));
    }

    @Override
    public void flatMap1(Stock stock, Collector<StockUpdate> out) throws Exception {
        Price price = priceState.value();
        if (price != null) {
            out.collect(StockUpdate.builder()
                    .symbol(stock.getSymbol())
                    .price(price.getPrice())
                    .companyName(stock.getCompanyName())
                    .timestamp(price.getTimestamp())
                    .build());
        }
    }

    @Override
    public void flatMap2(Price value, Collector<StockUpdate> out) throws Exception {
        priceState.update(value);
    }
}

@Slf4j
public class DataStreamJob {

    private static final Set<String> ALLOWED_STOCKS = Set.of("AAPL", "GOOG", "AMZN", "MSFT", "TSLA");

    public static void main(String[] args) throws Exception {
        // Sets up the execution environment, which is the main entry point to building Flink applications.
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Configure the environment
        configureEnvironment(env);

        // loading properties
        PropertiesLoader propertiesLoader = new PropertiesLoader();

        // Creating the execution plan
        DataStreamSource<Stock> stockDataStream = env.fromSource(SourceBuilder.stockSource(propertiesLoader.stockKafkaProperties()),
                WatermarkStrategy.forMonotonousTimestamps(),
                "stock-source");

        DataStreamSource<Price> priceDataStream = env.fromSource(SourceBuilder.priceSource(propertiesLoader.priceKafkaProperties()),
                WatermarkStrategy.forMonotonousTimestamps(),
                "price-source");

        // send the result to a kafka topic using KafkaSink
        KafkaSink<StockUpdate> stockUpdateKafkaSink = SourceBuilder.stockUpdateSink(propertiesLoader.stockUpdateKafkaProperties());

        stockDataStream
                .filter(stock -> ALLOWED_STOCKS.contains(stock.getSymbol()))
                .connect(priceDataStream)
                .keyBy(Stock::getSymbol, Price::getSymbol)
                .flatMap(new StockPriceJoiner())
                .name("join-stock-price")
                .map(stockUpdate -> {
                    log.info("stock price update sent to kafka: {}", stockUpdate);
                    return stockUpdate;
                })
                .sinkTo(stockUpdateKafkaSink);


        // Execute program, beginning computation.
        env.execute("StockPriceStreamJob");

    }

    private static void configureEnvironment(StreamExecutionEnvironment env) {
        // automatically recover from failures
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3, // number of restart attempts
                Time.seconds(10) // delay
        ));
        // disable kryo serialization and use the PojoSerializer (for better efficiency)
        env.getConfig().disableGenericTypes();
        // configuring the task manager
        env.configure(getConfiguration());
    }

    // Configure the task manager
    private static Configuration getConfiguration() {
        Configuration config = new Configuration();
        config.setString("taskmanager.cpu.cores", "4");
        config.setString("taskmanager.memory.task.heap.size", "1024m");
        return config;
    }
}