Deploying Flink Services

This section provides instructions for preparing persistent storage and deploying the CDP Flink Services on OpenShift.

Procedure

Prepare Flink persistent storage. Before deploying the Flink services, create the persistent volumes and persistent volume claims in the services deployment namespace. Replace the NFS path and server values with the values for your environment. Use the cdp-dev-app namespace and the storage class.

Create flink-conf-pvc.yaml with the following content and apply it to the deployment namespace.

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: flink-conf-pvc
  namespace: cdp-dev-app
spec:
  accessModes:
    - ReadWriteMany
  resources:
    requests:
      storage: 50Gi
  storageClassName: nfssc
  volumeMode: Filesystem
  volumeName: flink-conf-pv

kubectl apply -f flink-conf-pvc.yaml -n cdp-dev-app

Create flink-conf-pv.yaml and replace the NFS values with the values for your environment.

apiVersion: v1
kind: PersistentVolume
metadata:
  name: flink-conf-pv
spec:
  accessModes:
    - ReadWriteMany
  capacity:
    storage: 50Gi
  claimRef:
    apiVersion: v1
    kind: PersistentVolumeClaim
    name: flink-conf-pvc
    namespace: cdp-dev-app
  nfs:
    path: <NFS_PATH>/flink-conf
    server: <NFS_SERVER>
  persistentVolumeReclaimPolicy: Retain
  storageClassName: nfssc
  volumeMode: Filesystem

kubectl apply -f flink-conf-pv.yaml -n cdp-dev-app

Create flink-pvc.yaml and apply it to the deployment namespace.

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: flink-pvc
  namespace: cdp-dev-app
spec:
  accessModes:
    - ReadWriteMany
  resources:
    requests:
      storage: 50Gi
  storageClassName: nfssc
  volumeMode: Filesystem
  volumeName: flink-pv

kubectl apply -f flink-pvc.yaml -n cdp-dev-app

Create flink-pv.yaml and replace the NFS values with the values for your environment.

apiVersion: v1
kind: PersistentVolume
metadata:
  name: flink-pv
spec:
  accessModes:
    - ReadWriteMany
  capacity:
    storage: 50Gi
  claimRef:
    apiVersion: v1
    kind: PersistentVolumeClaim
    name: flink-pvc
    namespace: cdp-dev-app
  nfs:
    path: <NFS_PATH>/flink
    server: <NFS_SERVER>
  persistentVolumeReclaimPolicy: Retain
  storageClassName: nfssc
  volumeMode: Filesystem

kubectl apply -f flink-pv.yaml -n cdp-dev-app

Create flink-usrlib-pvc.yaml and apply it to the deployment namespace.

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: flink-usrlib-pvc
  namespace: cdp-dev-app
spec:
  accessModes:
    - ReadWriteMany
  resources:
    requests:
      storage: 50Gi
  storageClassName: nfssc
  volumeMode: Filesystem
  volumeName: flink-usrlib-pv

kubectl apply -f flink-usrlib-pvc.yaml -n cdp-dev-app

Create flink-usrlib-pv.yaml with the NFS path and server for the Flink user library, then apply it.

apiVersion: v1
kind: PersistentVolume
metadata:
  name: flink-usrlib-pv
spec:
  accessModes:
    - ReadWriteMany
  capacity:
    storage: 50Gi
  claimRef:
    apiVersion: v1
    kind: PersistentVolumeClaim
    name: flink-usrlib-pvc
    namespace: cdp-dev-app
  nfs:
    path: <NFS_PATH>/flink-conf
    server: <NFS_SERVER>
  persistentVolumeReclaimPolicy: Retain
  storageClassName: nfssc
  volumeMode: Filesystem

kubectl apply -f flink-usrlib-pv.yaml -n cdp-dev-app

Complete the prerequisites for CDP Flink Manager.

Create the cdp-flink-manager secret in HashiCorp Vault with the required values. Use the following keys and replace each placeholder with the value for your environment.

{
                        "DbName": "<database-name>",
                        "DbPassword": "<database-password>",
                        "DbUrl": "<database-url>",
                        "DbUsername": "<database-user>",
                        "KafkaBrokers": "<kafka-brokers>",
                        "KafkaPassword": "<kafka-password>",
                        "KafkaSaslMechanism": "<kafka-sasl-mechanism>",
                        "KafkaSecurityProtocol": "<kafka-security-protocol>",
                        "KafkaUsername": "<kafka-user>",
                        "cdpLoggingLevel": "INFO",
                        "flinkLoggingLevel": "WARN",
                        "kafkaLoggingLevel": "WARN",
                        "serverPort": "8181"
                        }

Update the ConfigMaps with the actual deployment values.
Create the required Kafka topics before deploying Flink Manager.

Deploy CDP Flink Manager.

In the Devtron Chart Store, select the cdp-flink-manager chart.
Configure the chart and select the option to deploy it.

In the YAML section, update the ConfigMap with the values for your environment and deploy the chart. Append tenant information to campaignDatabase in the format tenantId:segmentdb_tenantId.

AerospikePassword: <aerospike-password>
                    AerospikeServerAddress: <aerospike-server-address>
                    AerospikeUsername: <aerospike-username>
                    DbName: <database-name>
                    DbPassword: <database-password>
                    DbUrl: <database-url>
                    DbUsername: <database-username>
                    LOG_LEVEL_APACHE: INFO
                    LOG_LEVEL_APACHE_FLINK: INFO
                    LOG_LEVEL_CDP: INFO
                    LOG_LEVEL_ROOT: INFO
                    akkaAskTimeout: "600000"
                    autoscaleCheckInterval: "300000"
                    awsBucketName: <object-storage-bucket>
                    bucketType: minio
                    bufferTimeout: "100"
                    cacheRefreshIntervalMinutes: "3"
                    campaignDatabase: <tenant-id:segment-database-mappings>
                    cdpLoggingLevel: INFO
                    checkpointDirectory: flink/checkpoints
                    checkpointInterval: "180000"
                    checkpointMaxConcurrent: "1"
                    checkpointMinPauseBetween: "120000"
                    checkpointTimeout: "600000"
                    commitManifestMinCountToMerge: "10"
                    commitManifestTargetSizeBytes: "8388608"
                    connectionsMaxIdleMs: "540000"
                    defaultParallelism: "12"
                    deliveryTimeoutMs: "300000"
                    executionEngine: MINIO
                    flinkLoggingLevel: WARN
                    heartbeatInterval: "10000"
                    heartbeatTimeout: "30000"
                    hiveConfPath: <hive-configuration-path>
                    hiveMetastore: <hive-metastore-url>
                    icebergCheckpointInterval: "180000"
                    icebergParallelismDefault: "3"
                    icebergParallelismMax: "5"
                    icebergParallelismMin: "1"
                    icebergScaledownThreshold: "0.3"
                    icebergScaleupThreshold: "0.8"
                    icebergTableName: <iceberg-table-name>
                    icebergWriteDistributionMode: none
                    isAwsDevMode: "false"
                    jobmanagerHost: <job-manager-service-name>
                    jobmanagerPort: "8081"
                    kafkaBootstrapServers: <kafka-bootstrap-service>
                    kafkaConsumerFetchMessageMaxBytes: "52428800"
                    kafkaConsumerFetchMinBytes: "2097152"
                    kafkaConsumerFetchWaitMaxMs: "100"
                    kafkaConsumerHeartbeatIntervalMs: "8000"
                    kafkaConsumerMaxPartitionFetchBytes: "52428800"
                    kafkaConsumerMaxPollIntervalMs: "300000"
                    kafkaConsumerMaxPollRecords: "5000"
                    kafkaConsumerRebalanceTimeoutMs: "60000"
                    kafkaConsumerReceiveBufferBytes: "1048576"
                    kafkaConsumerRequestTimeoutMs: "120000"
                    kafkaConsumerSendBufferBytes: "1048576"
                    kafkaConsumerSessionTimeoutMs: "45000"
                    kafkaLoggingLevel: WARN
                    kafkaParallelismDefault: "3"
                    kafkaParallelismMax: "5"
                    kafkaParallelismMin: "1"
                    kafkaPassword: <kafka-password>
                    kafkaProducerAcks: all
                    kafkaProducerBatchSize: "1024288"
                    kafkaProducerBufferMemory: "134217728"
                    kafkaProducerCompressionType: lz4
                    kafkaProducerEnableIdempotence: "true"
                    kafkaProducerLingerMs: "200"
                    kafkaProducerMaxInFlightRequestsPerConnection: "5"
                    kafkaProducerMaxRequestSize: "10485760"
                    kafkaSaslMechanism: <kafka-sasl-mechanism>
                    kafkaScaledownThreshold: "0.3"
                    kafkaScaleupThreshold: "0.8"
                    kafkaSecurityProtocol: <kafka-security-protocol>
                    kafkaUsername: <kafka-username>
                    keyTypeCacheRefreshIntervalMinutes: "60"
                    metadataMaxAgeMs: "300000"
                    minioAccessKey: <minio-access-key>
                    minioCatalogName: <minio-catalog-name>
                    minioDatabaseMapping: <tenant-id:segment-database-mappings>
                    minioEnabled: "true"
                    minioEndpointURL: <minio-endpoint-url>
                    minioSecretKey: <minio-secret-key>
                    minioWarehouse: <minio-warehouse-url>
                    operatorChainingEnabled: "true"
                    orphanMessagesTopic: <failed-events-topic>
                    processingParallelism: "12"
                    reconciliationInterval: "300000"
                    region: <deployment-region>
                    requestTimeoutMs: "30000"
                    restartStrategyAttempts: "10"
                    restartStrategyDelay: "10000"
                    savepointDirectory: flink/savepoints
                    serverPort: "8181"
                    sinkFailureParallelism: "1"
                    sinkSuccessParallelism: "12"
                    slotRequestTimeout: "300000"
                    socketConnectionSetupTimeoutMaxMs: "30000"
                    socketConnectionSetupTimeoutMs: "10000"
                    sourceParallelism: "12"
                    taskCancellationTimeout: "300000"
                    transactionTimeoutMs: "900000"
                    trinoJdbcUrl: jdbc:trino://<trino-host>:<trino-port>/<catalog>?SSL=true&SSLVerification=NONE
                    trinoUsername: <trino-username>
                    warehouseSubpath: <warehouse-subpath>
                    webTimeout: "600000"
                    writeParquetRowGroupSizeBytes: "134217728"
                    writeTargetFileSizeBytes: "134217728"

Validate the deployment by checking the Flink Manager pods and deployment status.
```
oc get pods -n cdp-dev-app
oc get deployment -n cdp-dev-app
```