Deploying CDP Flink Manager

This section provides detailed instructions on how to deploy HCL CDP Flink Manager using the Devtron in the AWS..

Prerequisites:

Make sure to create cdp-flink-manager secret with required data in the AWS Secrets Manager, before you deploy Flink Manager.

To create the cdp-flink-manager secret in the AWS Secrets Manager, follow the steps below:

  1. Create a cdp-flink-manager secret sample key and value in the UI secret shown below, and update the actual values.
    {
  "DbUrl": ":3307",
  "DbName": "vrm",
  "DbUsername": "cdpapp_rw",
  "DbPassword": "",
  "KafkaBrokers": "",
  "KafkaSecurityProtocol": "SASL_SSL",
  "KafkaSaslMechanism": "SCRAM_SHA_512",
  "KafkaUsername": "",
  "KafkaPassword": "",
  "serverPort": "8181",
  "cdpLoggingLevel": "INFO",
  "flinkLoggingLevel": "WARN",
  "kafkaLoggingLevel": "WARN"
  2. Update ConfigMaps data with actual values, and create required kafka topics.

Deploying CDP Flink Manager

To deploy the CDP Flink Manager, follow the steps below:

  1. Navigate to the Devtron Chart Store, and select the cdp-flink-manager chart to deploy.

  2. Now, configure and deploy the charts.

  3. In the YAML section, update the ConfigMap using below details and deploy the charts.
    DbName: generic_db_name
DbPassword: "REDACTED_PASSWORD"
DbUrl: jdbc:mariadb://db-host-hostname:3307/generic_db_name?autoReconnect=true
DbUsername: generic_app_user
akkaAskTimeout: "600000"
bufferTimeout: "100"
cdpLoggingLevel: INFO
checkpointInterval: "180000"
checkpointMaxConcurrent: "1"
checkpointMinPauseBetween: "120000"
checkpointTimeout: "600000"
flinkLoggingLevel: WARN
heartbeatInterval: "10000"
heartbeatTimeout: "30000"
jobmanagerHost: job-manager-service-hostname
jobmanagerPort: "8081"
kafkaBootstrapServers: "kafka-broker-hostname:9092"
kafkaConsumerFetchMessageMaxBytes: "52428800"
kafkaConsumerFetchMinBytes: "2097152"
kafkaConsumerFetchWaitMaxMs: "100"
kafkaConsumerHeartbeatIntervalMs: "8000"
kafkaConsumerMaxPartitionFetchBytes: "52428800"
kafkaConsumerMaxPollIntervalMs: "300000"
kafkaConsumerMaxPollRecords: "5000"
kafkaConsumerReceiveBufferBytes: "1048576"
kafkaConsumerRequestTimeoutMs: "120000"
kafkaConsumerSendBufferBytes: "1048576"
kafkaConsumerSessionTimeoutMs: "45000"
kafkaConsumerRebalanceTimeoutMs: "60000"
kafkaLoggingLevel: WARN
kafkaPassword: REDACTED_KAFKA_PASSWORD
kafkaProducerAcks: all
kafkaProducerEnableIdempotence: "true"
kafkaProducerBatchSize: "1024288"
kafkaProducerBufferMemory: "134217728"
kafkaProducerCompressionType: lz4
kafkaProducerLingerMs: "200"
kafkaProducerMaxInFlightRequestsPerConnection: "5"
kafkaProducerMaxRequestSize: "10485760"
kafkaSaslMechanism: SCRAM_SHA_512
kafkaSecurityProtocol: SASL_SSL
kafkaUsername: generic_kafka_user
operatorChainingEnabled: "true"
orphanMessagesTopic: generic_failed_events_topic
restartStrategyAttempts: "10"
restartStrategyDelay: "10000"
serverPort: "8181"
kafkaParallelismDefault: "3"
icebergParallelismDefault: "3"
defaultParallelism: "12"
sourceParallelism: "12"
processingParallelism: "12"
sinkFailureParallelism: "1"
sinkSuccessParallelism: "12"
slotRequestTimeout: "300000"
taskCancellationTimeout: "300000"
webTimeout: "600000"
isAwsDevMode: "false"
icebergCheckpointInterval: "180000"
writeTargetFileSizeBytes: "134217728"
writeParquetRowGroupSizeBytes: "134217728"
icebergWriteDistributionMode: none
commitManifestTargetSizeBytes: "8388608"
commitManifestMinCountToMerge: "10"
awsAccessKey: YOUR_AWS_ACCESS_KEY
awsSecretKey: YOUR_AWS_SECRET_KEY
awsSessionToken: YOUR_AWS_SESSION_TOKEN
requestTimeoutMs: "30000"
deliveryTimeoutMs: "300000"
transactionTimeoutMs: "900000"
metadataMaxAgeMs: "300000"
connectionsMaxIdleMs: "540000"
socketConnectionSetupTimeoutMs: "10000"
socketConnectionSetupTimeoutMaxMs: "30000"
campaignDatabase: 0000:db_1,1111:db_2
icebergTableName: generic_table_name
cacheRefreshIntervalMinutes: "3"
keyTypeCacheRefreshIntervalMinutes: "60"
savepointDirectory: storage/savepoints
checkpointDirectory: storage/checkpoints
kafkaParallelismMin: "1"
kafkaParallelismMax: "5"
icebergParallelismMin: "1"
icebergParallelismMax: "5"
LOG_LEVEL_APACHE_FLINK: INFO
LOG_LEVEL_APACHE: INFO
LOG_LEVEL_CDP: INFO
LOG_LEVEL_ROOT: INFO
kafkaScaleupThreshold: "0.8"
kafkaScaledownThreshold: "0.3"
icebergScaleupThreshold: "0.8"
icebergScaledownThreshold: "0.3"
autoscaleCheckInterval: "300000"
reconciliationInterval: "300000"
AerospikeServerAddress: 0.0.0.0,0.0.0.1
AerospikePassword: REDACTED_AEROSPIKE_PASSWORD
AerospikeUsername: generic_aerospike_user
awsBucketName: your-s3-bucket-name
historyServerUrl: https://history-server.example.com
jobmanagerUrl: https://job-manager.example.com

  4. On successful deployment, validate the deployment as shown below.