Fundamentals and Core Concepts

# Topic configuration
num.partitions=12
replication.factor=3
min.insync.replicas=2
retention.ms=604800000  # 7 days
cleanup.policy=delete

// Flink streaming job example
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// Enable checkpointing for exactly-once processing
env.enableCheckpointing(60000); // 60 seconds
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// Configure Kafka source
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "kafka:9092");
properties.setProperty("group.id", "flink-consumer-group");

FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>(
    "input-topic",
    new SimpleStringSchema(),
    properties
);
kafkaSource.assignTimestampsAndWatermarks(
    WatermarkStrategy
        .<String>forBoundedOutOfOrderness(Duration.ofSeconds(5))
        .withTimestampAssigner((event, timestamp) -> extractTimestamp(event))
);

// Define the processing pipeline
DataStream<String> stream = env.addSource(kafkaSource);

// Parse JSON events
DataStream<Event> events = stream
    .map(json -> parseJson(json))
    .filter(event -> event.getType().equals("PURCHASE"));

// Process events with 5-minute tumbling windows
DataStream<Result> results = events
    .keyBy(event -> event.getUserId())
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new PurchaseAggregator());

// Write results to Kafka
FlinkKafkaProducer<String> kafkaSink = new FlinkKafkaProducer<>(
    "output-topic",
    new SimpleStringSchema(),
    properties
);

results
    .map(result -> convertToJson(result))
    .addSink(kafkaSink);

// Execute the job
env.execute("Purchase Processing Job");

// Kafka Streams application example
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "purchase-processor");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);

StreamsBuilder builder = new StreamsBuilder();

// Read from input topic
KStream<String, String> inputStream = builder.stream("input-topic");

// Parse JSON and filter events
KStream<String, Purchase> purchaseStream = inputStream
    .mapValues(value -> parsePurchase(value))
    .filter((key, purchase) -> purchase != null && purchase.getAmount() > 0);

// Group by user ID
KGroupedStream<String, Purchase> groupedByUser = purchaseStream
    .groupByKey(Grouped.with(Serdes.String(), purchaseSerde));

// Aggregate purchases in 5-minute windows
TimeWindows timeWindows = TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5));

KTable<Windowed<String>, UserPurchaseSummary> windowedAggregates = groupedByUser
    .windowedBy(timeWindows)
    .aggregate(
        UserPurchaseSummary::new,
        (userId, purchase, summary) -> summary.addPurchase(purchase),
        Materialized.<String, UserPurchaseSummary, WindowStore<Bytes, byte[]>>as("purchase-store")
            .withKeySerde(Serdes.String())
            .withValueSerde(userPurchaseSummarySerde)
    );

// Convert back to stream and write to output topic
windowedAggregates
    .toStream()
    .map((windowedKey, summary) -> KeyValue.pair(
        windowedKey.key(),
        formatSummaryAsJson(windowedKey.key(), summary, windowedKey.window().startTime())
    ))
    .to("output-topic", Produced.with(Serdes.String(), Serdes.String()));

// Build and start the topology
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();