flink-cluster.yaml

apiVersion: flinkoperator.k8s.io/v1beta1
kind: FlinkCluster
metadata:
  name: flink-cluster
  namespace: snp
spec:
  flinkProperties:
    kubernetes.cluster-id: snp-sn-platform
    kubernetes.namespace: snp
    taskmanager.numberOfTaskSlots: "1"
  image:
    name: streamnative/pulsar-flink:1.13.1.5-rc3-scala_2.12-gcp
    pullPolicy: IfNotPresent
  jobManager:
    accessScope: VPC
    memoryOffHeapMin: "0"
    memoryOffHeapRatio: 25
    ports:
      blob: 6124
      query: 6125
      rpc: 6123
      ui: 8081
    replicas: 1
    resources:
      limits:
        cpu: 500m
        memory: 1Gi
      requests:
        cpu: 500m
        memory: 1Gi
    volumeMounts:
    - mountPath: /opt/flink-sql-gateway/secret/pulsar-token/pulsar
      name: token-volume-pulsar
      readOnly: true
    volumes:
    - name: token-volume-pulsar
      secret:
        defaultMode: 420
        secretName: sql-auth
  taskManager:
    memoryOffHeapMin: "0"
    memoryOffHeapRatio: 25
    ports:
      data: 6121
      query: 6125
      rpc: 6122
    replicas: 4
    resources:
      limits:
        cpu: "500m"
        memory: 2Gi
      requests:
        cpu: "500m"
        memory: 2Gi
    volumeMounts:
    - mountPath: /opt/flink-sql-gateway/secret/pulsar-token/pulsar
      name: token-volume-pulsar
      readOnly: true
    volumes:
    - name: token-volume-pulsar
      secret:
        defaultMode: 420
        secretName: sql-auth
  logConfig:
    log4j-console.properties: |
      rootLogger.level = INFO
      rootLogger.appenderRef.file.ref = LogFile
      rootLogger.appenderRef.console.ref = LogConsole
      appender.file.name = LogFile
      appender.file.type = File
      appender.file.append = false
      appender.file.fileName = ${sys:log.file}
      appender.file.layout.type = PatternLayout
      appender.file.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
      appender.console.name = LogConsole
      appender.console.type = CONSOLE
      appender.console.layout.type = PatternLayout
      appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
      logger.akka.name = akka
      logger.akka.level = INFO
      logger.kafka.name= org.apache.kafka
      logger.kafka.level = INFO
      logger.hadoop.name = org.apache.hadoop
      logger.hadoop.level = INFO
      logger.zookeeper.name = org.apache.zookeeper
      logger.zookeeper.level = INFO
      logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
      logger.netty.level = OFF
    logback-console.xml: |
      <configuration>
        <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
          <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{60} %X{sourceThread} - %msg%n</pattern>
          </encoder>
        </appender>
        <appender name="file" class="ch.qos.logback.core.FileAppender">
          <file>${log.file}</file>
          <append>false</append>
          <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{60} %X{sourceThread} - %msg%n</pattern>
          </encoder>
        </appender>
        <root level="DEBUG">
          <appender-ref ref="console"/>
          <appender-ref ref="file"/>
        </root>
        <logger name="akka" level="INFO" />
        <logger name="org.apache.kafka" level="INFO" />
        <logger name="org.apache.hadoop" level="INFO" />
        <logger name="org.apache.zookeeper" level="INFO" />
        <logger name="org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline" level="ERROR" />
      </configuration>
  recreateOnUpdate: true