Developing Applications On Managed Service For Apache Flink

docker --version
docker-compose --version
docker run hello-world

java -version
javac -version
echo $JAVA_HOME

version: '3.8'

# Set FLINK_IMAGE_TAG in your environment or .env file:
#   Flink 1.20: FLINK_IMAGE_TAG=1.20-java11
#   Flink 2.2:  FLINK_IMAGE_TAG=2.2-java17

services:
  # Flink JobManager
  jobmanager:
    image: flink:${FLINK_IMAGE_TAG:-2.2-java17}
    hostname: jobmanager
    container_name: flink-jobmanager
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        jobmanager.memory.process.size: 1600m
        jobmanager.execution.failover-strategy: region
      - RUNTIME_ENVIRONMENT=local
    volumes:
      - ./flink-jobs:/opt/flink/jobs
      - flink-checkpoints:/tmp/flink-checkpoints
      - flink-savepoints:/tmp/flink-savepoints
    networks:
      - flink-network

  # Flink TaskManager
  taskmanager:
    image: flink:${FLINK_IMAGE_TAG:-2.2-java17}
    depends_on:
      - jobmanager
    command: taskmanager
    scale: 2
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        taskmanager.numberOfTaskSlots: 2
        parallelism.default: 2        
        taskmanager.memory.process.size: 1728m
        taskmanager.memory.managed.fraction: 0.4
        state.backend: rocksdb
        state.checkpoints.dir: file:///tmp/flink-checkpoints
        state.savepoints.dir: file:///tmp/flink-savepoints
        execution.checkpointing.interval: 60000
        execution.checkpointing.mode: EXACTLY_ONCE
        restart-strategy: exponential-delay
        restart-strategy.exponential-delay.initial-backoff: 10s
        restart-strategy.exponential-delay.max-backoff: 2min
        restart-strategy.exponential-delay.backoff-multiplier: 2.0
      - RUNTIME_ENVIRONMENT=local
    volumes:
      - ./flink-jobs:/opt/flink/jobs
      - flink-checkpoints:/tmp/flink-checkpoints
      - flink-savepoints:/tmp/flink-savepoints
    networks:
      - flink-network

  # Kafka (for streaming data sources)
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    networks:
      - flink-network

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    hostname: kafka
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "9101:9101"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
    networks:
      - flink-network

  # LocalStack (for local AWS service emulation: Kinesis, S3, DynamoDB, etc.)
  localstack:
    image: localstack/localstack:3.5.0
    container_name: localstack
    ports:
      - "4566:4566"
    environment:
      SERVICES: kinesis,s3,dynamodb,cloudwatch
      DEFAULT_REGION: us-east-1
    networks:
      - flink-network

volumes:
  flink-checkpoints:
  flink-savepoints:

networks:
  flink-network:
    driver: bridge

# Create Managed Service for Apache Flink project directory
mkdir my-msf-app
cd my-msf-app

# Create required directories
mkdir -p flink-jobs src/main/java src/test/java

# Copy docker-compose.yml to project root

# Start all services
docker-compose up -d

# Verify services are running
docker-compose ps

# Check taskmanager logs for submitted Flink jobs 
docker-compose logs taskmanager

# Check Flink Web UI
open http://localhost:8081

# Check Kafka
docker-compose exec kafka kafka-topics --bootstrap-server localhost:9092 --list

# For Flink 1.20 (Java 11):
mvn clean package -Plocal -q

# For Flink 2.2 (Java 17):
# Set JAVA_HOME to your Java 17 installation path before building
export JAVA_HOME=<path-to-java-17>
mvn clean package -Plocal -q

cp target/my-flink-app-1.0.jar flink-jobs/

docker exec flink-jobmanager flink run /opt/flink/jobs/my-flink-app-1.0.jar

# List Managed Service for Apache Flink applications
aws kinesisanalyticsv2 list-applications

# Describe Managed Service for Apache Flink application
aws kinesisanalyticsv2 describe-application --application-name my-app

# Create savepoint
aws kinesisanalyticsv2 create-application-snapshot --application-name my-app --snapshot-name my-snapshot