SourceFunction

DiscardingSink

SourceFunction

Source

SingleThreadMultiplexSourceReaderBase

podTemplate

apiVersion: <http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>
kind: FlinkDeployment
metadata:
  name: flink01
spec:
  image: flink01:1
  flinkVersion: v1_15
  podTemplate:
    spec:
      containers:
        - name: flink-main-container
          volumeMounts:
            - mountPath: /opt/flink/log
              name: flink-logs
          env:
            - name: MY_CUSTOM_ENV
              value: "SUPER"
      volumes:
        - name: flink-logs
          emptyDir: { }
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
  serviceAccount: flink
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "2048m"
      cpu: 1
  job:
    jarURI: local:///opt/flink/usrlib/my.jar
    parallelism: 2
    upgradeMode: stateless

[CL] [WindowOperator] Initializing internalTimerService with wm -9223372036854775808
[CL] [InternalWindowProcessFunction] this.ctx.currentWatermark() at open(): -9223372036854775808
[CL] Opening ProcTimeDeduplicateKeepLastRowFunction
[CL] [ProcTimeDedupeFn] cw -9223372036854775808 for input org.apache.flink.table.data.binary.BinaryRowData@f17bddd6

2022-09-23 08:36:15,992 WARN  org.apache.hadoop.fs.s3a.S3AInstrumentation                  [] - Closing output stream statistics while data is still marked as pending upload in OutputStreamStatistics{blocksSubmitted=1, blocksInQueue=1, blocksActive=0, blockUploadsCompleted=0, blockUploadsFailed=0, bytesPendingUpload=9759410, bytesUploaded=0, blocksAllocated=1, blocksReleased=1, blocksActivelyAllocated=0, exceptionsInMultipartFinalize=0, transferDuration=0 ms, queueDuration=0 ms, averageQueueTime=0 ms, totalUploadDuration=0 ms, effectiveBandwidth=0.0 bytes/s}

java.util.concurrent.CancellationException: null

org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable [] - VideoQos Stream Reporter (14/40)#0 - asynchronous part of checkpoint 5149 could not be completed.

org.apache.flink.runtime.checkpoint.CheckpointCoordinator    [] - Checkpoint 5149 of job 22c1b49deb67c5388647ac9a79bf87ca expired before completing.

kubernetes.rest-service.exposed.type

################################################################################
#  Licensed to the Apache Software Foundation (ASF) under one
#  or more contributor license agreements.  See the NOTICE file
#  distributed with this work for additional information
#  regarding copyright ownership.  The ASF licenses this file
#  to you under the Apache License, Version 2.0 (the
#  "License"); you may not use this file except in compliance
#  with the License.  You may obtain a copy of the License at
#
#      <http://www.apache.org/licenses/LICENSE-2.0>
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
# limitations under the License.
################################################################################

apiVersion: <http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>
kind: FlinkDeployment
metadata:
  name: statemachineflink
spec:
  image: flink:1.15
  flinkVersion: v1_15
  ingress:
    template: "xxx"
    className: "alb"
  flinkConfiguration:
    kubernetes.rest-service.exposed.type: "Loadbalancer"
    containerized.master.env.ENABLE_BUILT_IN_PLUGINS: "flink-s3-fs-presto-1.15.2.jar"
    containerized.taskmanager.env.ENABLE_BUILT_IN_PLUGINS: "flink-s3-fs-presto-1.15.2.jar"
    taskmanager.numberOfTaskSlots: "2"
    state.savepoints.dir: "<s3://xxx>"
    state.checkpoints.dir: "<s3://xxx>"
    high-availability: "org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory"
    high-availability.storageDir: "xxx"
    execution.checkpointing.interval: 10s
    state.backend: rocksdb
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "2048m"
      cpu: 1
  podTemplate:
    metadata:
      labels:
        app: statemachineflink
    spec:
      containers:
        - name: flink-main-container
          volumeMounts:
          - mountPath: /flink-data
            name: flink-volume
      volumes:
        - name: flink-volume
          emptyDir: {}
  job:
    jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
    parallelism: 2
    upgradeMode: savepoint
    state: running
    savepointTriggerNonce: 0

yarn-per-job

SUSPENDED

Client session timed out, have not heard from server in 40019ms for sessionid 0x20a20d1641b67b4

org.apache.flink.runtime.dispatcher.MiniDispatcher           [] - Shutting down cluster with state SUSPENDED, jobCancelled: false, executionMode: DETACHED
org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - Shutting YarnJobClusterEntrypoint down with application status UNKNOWN. Diagnostics null.

org.apache.flink.runtime.checkpoint.DefaultCompletedCheckpointStore [] - Recovering checkpoints from ZooKeeperStateHandleStore{namespace='flink/application_1661494717285_98085/checkpoints/a636bd043052b707aadb1a7b98fe99d5'}.
org.apache.flink.runtime.checkpoint.DefaultCompletedCheckpointStore [] - Found 0 checkpoints in ZooKeeperStateHandleStore{namespace='flink/application_1661494717285_98085/checkpoints/a636bd043052b707aadb1a7b98fe99d5'}.

org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperHaServices [] - Close and clean up all data for ZooKeeperHaServices.
org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperHaServices [] - Finished cleaning up the high availability data.

Sync duration

The duration of the synchronous part of the checkpoint. This includes snapshotting state of the operators and blocks all other activity on the subtask (processing records, firing timers, etc).

Subsequently, the accumulate(...) method of the function is called for each input row to update the accumulator. Once all rows have been processed, the getValue(...) method of the function is called to compute and return the final result.

rows have been processed

getValue()

accumulate()

val env = StreamExecutionEnvironment.getExecutionEnvironment()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    env.enableCheckpointing(5000)
    env.checkpointConfig.setCheckpointStorage("file:///Users/xyz/dev/my-flink-app/checkpoints")
    env.stateBackend = EmbeddedRocksDBStateBackend(true)
    
    // ...
    
    env.execute("Stream Join Demo"

gs

<property>
    <name>fs.gs.impl</name>
    <value>org.apache.flink.fs.gs.GSFileSystem</value>
  </property>

root@airstream-f69f865f8-xcqfh:/opt/flink# ls plugins/gs-fs-hadoop/
flink-gs-fs-hadoop-1.15.2.jar
root@airstream-f69f865f8-xcqfh:/opt/flink# ls opt/flink-gs-fs-hadoop-1.15.2.jar
opt/flink-gs-fs-hadoop-1.15.2.jar

flink-kubernetes-operator

Error: failed to download "helm/flink-kubernetes-operator" (hint: running `helm repo update` may help)

helm upgrade --install --wait flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator --set watchNamespaces={ns1,ns2}

I see that flink operator pod not getting restarted after helm upgrade.

Hi All,

We are trying to enrich streaming data with a look up data. Look up data will be kind of static(updated once in a while).
For our streaming application we are using flink table API. To handle this scenario we are trying to use flink temporal join.
We have create two tables for our sources as below (below example is mocked one)


1. Look up data

	CREATE TABLE versioned_rates (
		currency STRING,
		rate DECIMAL(38, 10),
		currency_time TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
		WATERMARK FOR currency_time AS currency_time,
		PRIMARY KEY(currency) NOT ENFORCED 
	) WITH (
	    'connector' = 'upsert-kafka',
	    ...)

2. Streaming data (source streaming data)
	
	CREATE TABLE orders (
	    order_id STRING,
	    currency STRING,
	    amount INT,
	    order_time TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
	    WATERMARK FOR order_time AS order_time
	) WITH (
	    'connector' = 'kafka',
	    ...)


The join on these tables is like below

	SELECT 
	  o.order_id,
	  o.order_time,
	  o.amount * r.rate AS amount,
	  r.rate rate,
	  r.currency
	FROM orders AS o JOIN versioned_rates FOR SYSTEM_TIME AS OF o.order_time r
	on o.currency = r.currency;

The issue we are facing is that we are not getting output on this join if there is no update in the look up data.

We need the behaviour like, the data should be written for every record we receive in streaming data even if there is no update in look up data

After doing some studies we understand that this may be related to watermark is not getting matched between streams,
but we couldn't clearly understand why. If we can get any help on this that would be appreciated.

All important decisions and conclusions *must be reflected back to the mailing lists.* "If it didn't happen on a mailing list, it didn't happen." - The Apache Mottos

logs)

users

.withTimestampAssigner(...)

FlinkKafkaConsumer

DataStream

Table

log's

users

No Watermark (Watermarks are only available if EventTime is used)

users

logs

users

// Initialize Flink job environment. Some properties are omitted for readability ...
StreamExecutionEnvironment env = ...
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
tableEnv.getConfig().setLocalTimeZone(ZoneId.of("UTC"));
...

/* 1.1 Read 'users' Kafka topic*/
DataStream<UserDto> usersStream = env
        .addSource(new CustomFlinkKafkaConsumer<>(...))
        .uid("source-users").name("Read Users")
        .map(UserRecordMapper::mapToUserDto)
        .uid("map-to-user-model")
        .keyBy(UserDto::getUId);

// 1.2 Datastream to Table
Table tUsers = tableEnv.fromDataStream(usersStream, Schema.newBuilder()
        .primaryKey("u_id")
        .columnByExpression("lastModifiedTs2", "TO_TIMESTAMP_LTZ(lastModifiedTs, 3)")
        .watermark("lastModifiedTs2", "lastModifiedTs2 - interval '1' day")
        .build());
tUsers.printSchema();
tableEnv.createTemporaryView("users", tUsers);

logs

/* 2.1) Read 'logs' Kafka topic*/
DataStream<LoggingAudit> logAuditStream = env
        .addSource(new CustomFlinkKafkaConsumer<>(...))
        .uid("source-logs").name("Read Logs")
        .filter(l -> Objects.nonNull(l.getUserId()))
        .keyBy(LoggingAudit::getUserId);

DataStream<LoggingAuditDto> transformedLogAudit = logAuditStream
        .map(new LogAuditTransformer())
        .uid("transform-log-audit").name("Transform Log Audit");

// 2.2) Datastream to Table
Schema logAuditTableSchema = Schema.newBuilder()
        .columnByExpression("insertDateTs2", "TO_TIMESTAMP_LTZ(insertDateTs, 3)")
        .watermark("insertDateTs2", "insertDateTs2 - interval '1' day")
        .build();
Table logAuditTable = tableEnv.fromDataStream(transformedLogAudit, logAuditTableSchema);
logAuditTable.printSchema();
tableEnv.createTemporaryView("logs", logAuditTable);

tableEnv.sqlQuery("select logs.id, logs.action, users.u_id as userId " +
                "from logs " +
                "join users FOR SYSTEM_TIME AS OF logs.insertDateTs2 " +
                "on logs.userId = users.u_id")
        .execute()
        .print();

assignTimestampsAndWatermarks

jobmanager.memory.process.size: "1500m"

taskmanager.memory.process.size: "1500m"

taskmanager.memory.jvm-metaspace.size: "256m"

cpuCores=1.0, taskHeapSize=270.000mb (283115513 bytes), taskOffHeapSize=0 bytes, networkMemSize=105.200mb (110310196 bytes), managedMemSize=420.800mb (441240787 bytes), numSlots=2

taskOffHeapSize=0