You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2021/04/01 02:22:21 UTC
[flink] branch master updated: [FLINK-21191][upsert-kafka] Support
buffered sink function for upsert-kafka
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new ec9b0c5 [FLINK-21191][upsert-kafka] Support buffered sink function for upsert-kafka
ec9b0c5 is described below
commit ec9b0c5b60290697769415eb3e1b1ed2052460ac
Author: Shengkai <10...@qq.com>
AuthorDate: Tue Mar 30 22:40:31 2021 +0800
[FLINK-21191][upsert-kafka] Support buffered sink function for upsert-kafka
This closes #15434
---
.../docs/connectors/table/upsert-kafka.md | 17 ++
docs/content/docs/connectors/table/upsert-kafka.md | 22 ++
.../kafka/table/BufferedUpsertSinkFunction.java | 313 +++++++++++++++++++
.../table/DynamicKafkaSerializationSchema.java | 2 +-
.../connectors/kafka/table/KafkaDynamicSink.java | 56 +++-
.../kafka/table/KafkaDynamicTableFactory.java | 2 +
.../connectors/kafka/table/KafkaOptions.java | 26 ++
.../kafka/table/SinkBufferFlushMode.java | 79 +++++
.../table/UpsertKafkaDynamicTableFactory.java | 44 ++-
.../table/BufferedUpsertSinkFunctionTest.java | 330 +++++++++++++++++++++
.../kafka/table/KafkaDynamicTableFactoryTest.java | 2 +
.../table/UpsertKafkaDynamicTableFactoryTest.java | 94 ++++--
12 files changed, 951 insertions(+), 36 deletions(-)
diff --git a/docs/content.zh/docs/connectors/table/upsert-kafka.md b/docs/content.zh/docs/connectors/table/upsert-kafka.md
index ea4dc64..d269e67 100644
--- a/docs/content.zh/docs/connectors/table/upsert-kafka.md
+++ b/docs/content.zh/docs/connectors/table/upsert-kafka.md
@@ -185,6 +185,23 @@ of all available metadata fields.
<td>Integer</td>
<td>定义 upsert-kafka sink 算子的并行度。默认情况下，由框架确定并行度，与上游链接算子的并行度保持一致。</td>
</tr>
+ <tr>
+ <td><h5>sink.buffer-flush.max-rows</h5></td>
+ <td>可选</td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Integer</td>
+ <td>缓存刷新前，最多能缓存多少条记录。当 sink 收到很多同 key 上的更新时，缓存将保留同 key 的最后一条记录，因此 sink 缓存能帮助减少发往 Kafka topic 的数据量，以及避免发送潜在的 tombstone 消息。
+ 可以通过设置为 '0' 来禁用它。默认，该选项是未开启的。注意，如果要开启 sink 缓存，需要同时设置 <code>'sink.buffer-flush.max-rows'</code>
+ 和 <code>'sink.buffer-flush.interval'</code> 两个选项为大于零的值。</td>
+ </tr>
+ <tr>
+ <td><h5>sink.buffer-flush.interval</h5></td>
+ <td>可选</td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Duration</td>
+ <td>缓存刷新的间隔时间，超过该时间后异步线程将刷新缓存数据。当 sink 收到很多同 key 上的更新时，缓存将保留同 key 的最后一条记录，因此 sink 缓存能帮助减少发往 Kafka topic 的数据量，以及避免发送潜在的 tombstone 消息。
+ 可以通过设置为 '0' 来禁用它。默认，该选项是未开启的。注意，如果要开启 sink 缓存，需要同时设置 <code>'sink.buffer-flush.max-rows'</code>
+ 和 <code>'sink.buffer-flush.interval'</code> 两个选项为大于零的值。</td>
</tbody>
</table>
diff --git a/docs/content/docs/connectors/table/upsert-kafka.md b/docs/content/docs/connectors/table/upsert-kafka.md
index 6bf75e9..03d87da 100644
--- a/docs/content/docs/connectors/table/upsert-kafka.md
+++ b/docs/content/docs/connectors/table/upsert-kafka.md
@@ -196,6 +196,28 @@ Connector Options
<td>Integer</td>
<td>Defines the parallelism of the upsert-kafka sink operator. By default, the parallelism is determined by the framework using the same parallelism of the upstream chained operator.</td>
</tr>
+ <tr>
+ <td><h5>sink.buffer-flush.max-rows</h5></td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Integer</td>
+ <td>The max size of buffered records before flush.
+ When the sink receives many updates on the same key, the buffer will retain the last record of the same key.
+ This can help to reduce data shuffling and avoid possible tombstone messages to Kafka topic. Can be set to '0' to disable it.
+ By default, this is disabled. Note both <code>'sink.buffer-flush.max-rows'</code> and
+ <code>'sink.buffer-flush.interval'</code> must be set to be greater than zero to enable sink buffer flushing.</td>
+ </tr>
+ <tr>
+ <td><h5>sink.buffer-flush.interval</h5></td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Duration</td>
+ <td>The flush interval mills, over this time, asynchronous threads will flush data.
+ When the sink receives many updates on the same key, the buffer will retain the last record of the same key.
+ This can help to reduce data shuffling and avoid possible tombstone messages to Kafka topic. Can be set to '0' to disable it.
+ By default, this is disabled. Note both <code>'sink.buffer-flush.max-rows'</code> and
+ <code>'sink.buffer-flush.interval'</code> must be set to be greater than zero to enable sink buffer flushing.</td>
+ </tr>
</tbody>
</table>
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/BufferedUpsertSinkFunction.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/BufferedUpsertSinkFunction.java
new file mode 100644
index 0000000..4c2c455
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/BufferedUpsertSinkFunction.java
@@ -0,0 +1,313 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.streaming.connectors.kafka.table;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.state.CheckpointListener;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.runtime.util.ExecutorThreadFactory;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.types.RowKind;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import static org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaSerializationSchema.createProjectedRow;
+import static org.apache.flink.types.RowKind.DELETE;
+import static org.apache.flink.types.RowKind.UPDATE_AFTER;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The wrapper of the {@link RichSinkFunction}. It will buffer the data and emit when the buffer is
+ * full or timer is triggered or checkpointing.
+ */
+public class BufferedUpsertSinkFunction extends RichSinkFunction<RowData>
+ implements CheckpointedFunction, CheckpointListener {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(BufferedUpsertSinkFunction.class);
+
+ // --------------------------------------------------------------------------------------------
+ // Config
+ // --------------------------------------------------------------------------------------------
+
+ private final RichSinkFunction<RowData> producer;
+ private final int batchMaxRowNums;
+ private final long batchIntervalMs;
+ private final DataType physicalDataType;
+ private final int[] keyProjection;
+ private final TypeInformation<RowData> consumedRowDataTypeInfo;
+ private boolean closed;
+
+ // --------------------------------------------------------------------------------------------
+ // Writer and buffer
+ // --------------------------------------------------------------------------------------------
+
+ private int batchCount = 0;
+ private transient Map<RowData, Tuple2<RowData, Long>> reduceBuffer;
+ private transient WrappedContext wrappedContext;
+ private transient Function<RowData, RowData> keyExtractor;
+ private transient Function<RowData, RowData> valueCopier;
+
+ // --------------------------------------------------------------------------------------------
+ // Timer attributes
+ // --------------------------------------------------------------------------------------------
+
+ private transient ScheduledExecutorService scheduler;
+ private transient ScheduledFuture<?> scheduledFuture;
+ private transient volatile Exception flushException;
+
+ public BufferedUpsertSinkFunction(
+ RichSinkFunction<RowData> producer,
+ DataType physicalDataType,
+ int[] keyProjection,
+ TypeInformation<RowData> consumedRowDataTypeInfo,
+ SinkBufferFlushMode bufferFlushMode) {
+ checkArgument(bufferFlushMode != null && bufferFlushMode.isEnabled());
+ this.producer = checkNotNull(producer, "Producer must not be null.");
+ this.physicalDataType =
+ checkNotNull(physicalDataType, "Physical data type must not be null.");
+ this.keyProjection = checkNotNull(keyProjection, "key projection must not be null.");
+ this.consumedRowDataTypeInfo = consumedRowDataTypeInfo;
+ this.batchMaxRowNums = bufferFlushMode.getBatchSize();
+ this.batchIntervalMs = bufferFlushMode.getBatchIntervalMs();
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ // init variable
+ reduceBuffer = new HashMap<>();
+ wrappedContext = new WrappedContext();
+ closed = false;
+
+ // create keyExtractor and value copier
+ List<LogicalType> fields = physicalDataType.getLogicalType().getChildren();
+ final RowData.FieldGetter[] keyFieldGetters =
+ Arrays.stream(keyProjection)
+ .mapToObj(
+ targetField ->
+ RowData.createFieldGetter(
+ fields.get(targetField), targetField))
+ .toArray(RowData.FieldGetter[]::new);
+ this.keyExtractor = rowData -> createProjectedRow(rowData, RowKind.INSERT, keyFieldGetters);
+
+ TypeSerializer<RowData> typeSerializer =
+ consumedRowDataTypeInfo.createSerializer(getRuntimeContext().getExecutionConfig());
+ this.valueCopier =
+ getRuntimeContext().getExecutionConfig().isObjectReuseEnabled()
+ ? Function.identity()
+ : typeSerializer::copy;
+
+ // register timer
+ this.scheduler =
+ Executors.newScheduledThreadPool(
+ 1, new ExecutorThreadFactory("upsert-kafka-sink-function"));
+ this.scheduledFuture =
+ this.scheduler.scheduleWithFixedDelay(
+ () -> {
+ synchronized (BufferedUpsertSinkFunction.this) {
+ if (!closed) {
+ try {
+ flush();
+ } catch (Exception e) {
+ flushException = e;
+ }
+ }
+ }
+ },
+ batchIntervalMs,
+ batchIntervalMs,
+ TimeUnit.MILLISECONDS);
+
+ producer.open(parameters);
+ }
+
+ @Override
+ public void setRuntimeContext(RuntimeContext t) {
+ producer.setRuntimeContext(t);
+ }
+
+ @Override
+ public RuntimeContext getRuntimeContext() {
+ return producer.getRuntimeContext();
+ }
+
+ @Override
+ public void invoke(RowData value, Context context) throws Exception {
+ wrappedContext.setContext(context);
+ addToBuffer(value, context.timestamp());
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ if (producer instanceof CheckpointListener) {
+ ((CheckpointListener) producer).notifyCheckpointComplete(checkpointId);
+ }
+ }
+
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) throws Exception {
+ if (producer instanceof CheckpointListener) {
+ ((CheckpointListener) producer).notifyCheckpointAborted(checkpointId);
+ }
+ }
+
+ @Override
+ public synchronized void close() throws Exception {
+ if (!closed) {
+ closed = true;
+
+ if (this.scheduledFuture != null) {
+ scheduledFuture.cancel(false);
+ this.scheduler.shutdown();
+ }
+
+ if (batchCount > 0) {
+ try {
+ flush();
+ } catch (Exception e) {
+ LOG.warn("Writing records to kafka failed.", e);
+ throw new RuntimeException("Writing records to kafka failed.", e);
+ }
+ }
+
+ producer.close();
+ }
+ super.close();
+ checkFlushException();
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ flush();
+ if (producer instanceof CheckpointedFunction) {
+ ((CheckpointedFunction) producer).snapshotState(context);
+ }
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ if (producer instanceof CheckpointedFunction) {
+ ((CheckpointedFunction) producer).initializeState(context);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private synchronized void addToBuffer(RowData row, Long timestamp) throws Exception {
+ checkFlushException();
+
+ RowData key = keyExtractor.apply(row);
+ RowData value = valueCopier.apply(row);
+ reduceBuffer.put(key, new Tuple2<>(changeFlag(value), timestamp));
+ batchCount++;
+
+ if (batchCount >= batchMaxRowNums) {
+ flush();
+ }
+ }
+
+ private synchronized void flush() throws Exception {
+ checkFlushException();
+ for (Tuple2<RowData, Long> value : reduceBuffer.values()) {
+ wrappedContext.setTimestamp(value.f1);
+ producer.invoke(value.f0, wrappedContext);
+ }
+ reduceBuffer.clear();
+ batchCount = 0;
+ }
+
+ private RowData changeFlag(RowData value) {
+ switch (value.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ value.setRowKind(UPDATE_AFTER);
+ break;
+ case UPDATE_BEFORE:
+ case DELETE:
+ value.setRowKind(DELETE);
+ }
+ return value;
+ }
+
+ private void checkFlushException() {
+ if (flushException != null) {
+ throw new RuntimeException("Writing records to JDBC failed.", flushException);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Wrapper of {@link Context}.
+ *
+ * <p>When records arrives, the {@link BufferedUpsertSinkFunction} updates the current {@link
+ * Context} and memorize the timestamp with the records. When flushing, the {@link
+ * BufferedUpsertSinkFunction} will emit the records in the buffer with memorized timestamp.
+ */
+ private static class WrappedContext implements Context {
+
+ private Long timestamp;
+ private Context context;
+
+ void setTimestamp(Long timestamp) {
+ this.timestamp = timestamp;
+ }
+
+ void setContext(Context context) {
+ this.context = context;
+ }
+
+ @Override
+ public long currentProcessingTime() {
+ return context.currentProcessingTime();
+ }
+
+ @Override
+ public long currentWatermark() {
+ return context.currentWatermark();
+ }
+
+ @Override
+ public Long timestamp() {
+ return timestamp;
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaSerializationSchema.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaSerializationSchema.java
index c4b6776..b2a0b21 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaSerializationSchema.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaSerializationSchema.java
@@ -187,7 +187,7 @@ class DynamicKafkaSerializationSchema
return null;
}
- private static RowData createProjectedRow(
+ static RowData createProjectedRow(
RowData consumedRow, RowKind kind, RowData.FieldGetter[] fieldGetters) {
final int arity = fieldGetters.length;
final GenericRowData genericRowData = new GenericRowData(kind, arity);
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java
index 29db659..f0d0a48 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java
@@ -35,7 +35,6 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.utils.DataTypeUtils;
-import org.apache.flink.util.Preconditions;
import org.apache.kafka.common.header.Header;
@@ -51,6 +50,8 @@ import java.util.Objects;
import java.util.Properties;
import java.util.stream.Stream;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
/** A version-agnostic Kafka {@link DynamicTableSink}. */
@Internal
public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetadata {
@@ -66,6 +67,9 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada
// Format attributes
// --------------------------------------------------------------------------------------------
+ /** Data type of consumed data type. */
+ protected DataType consumedDataType;
+
/** Data type to configure the formats. */
protected final DataType physicalDataType;
@@ -106,10 +110,14 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada
*/
protected final boolean upsertMode;
+ /** Sink buffer flush config which only supported in upsert mode now. */
+ protected final SinkBufferFlushMode flushMode;
+
/** Parallelism of the physical Kafka producer. * */
protected final @Nullable Integer parallelism;
public KafkaDynamicSink(
+ DataType consumedDataType,
DataType physicalDataType,
@Nullable EncodingFormat<SerializationSchema<RowData>> keyEncodingFormat,
EncodingFormat<SerializationSchema<RowData>> valueEncodingFormat,
@@ -121,28 +129,32 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada
@Nullable FlinkKafkaPartitioner<RowData> partitioner,
KafkaSinkSemantic semantic,
boolean upsertMode,
+ SinkBufferFlushMode flushMode,
@Nullable Integer parallelism) {
// Format attributes
+ this.consumedDataType =
+ checkNotNull(consumedDataType, "Consumed data type must not be null.");
this.physicalDataType =
- Preconditions.checkNotNull(
- physicalDataType, "Physical data type must not be null.");
+ checkNotNull(physicalDataType, "Physical data type must not be null.");
this.keyEncodingFormat = keyEncodingFormat;
this.valueEncodingFormat =
- Preconditions.checkNotNull(
- valueEncodingFormat, "Value encoding format must not be null.");
- this.keyProjection =
- Preconditions.checkNotNull(keyProjection, "Key projection must not be null.");
- this.valueProjection =
- Preconditions.checkNotNull(valueProjection, "Value projection must not be null.");
+ checkNotNull(valueEncodingFormat, "Value encoding format must not be null.");
+ this.keyProjection = checkNotNull(keyProjection, "Key projection must not be null.");
+ this.valueProjection = checkNotNull(valueProjection, "Value projection must not be null.");
this.keyPrefix = keyPrefix;
// Mutable attributes
this.metadataKeys = Collections.emptyList();
// Kafka-specific attributes
- this.topic = Preconditions.checkNotNull(topic, "Topic must not be null.");
- this.properties = Preconditions.checkNotNull(properties, "Properties must not be null.");
+ this.topic = checkNotNull(topic, "Topic must not be null.");
+ this.properties = checkNotNull(properties, "Properties must not be null.");
this.partitioner = partitioner;
- this.semantic = Preconditions.checkNotNull(semantic, "Semantic must not be null.");
+ this.semantic = checkNotNull(semantic, "Semantic must not be null.");
this.upsertMode = upsertMode;
+ this.flushMode = checkNotNull(flushMode);
+ if (flushMode.isEnabled() && !upsertMode) {
+ throw new IllegalArgumentException(
+ "Sink buffer flush is only supported in upsert-kafka.");
+ }
this.parallelism = parallelism;
}
@@ -162,7 +174,18 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada
final FlinkKafkaProducer<RowData> kafkaProducer =
createKafkaProducer(keySerialization, valueSerialization);
- return SinkFunctionProvider.of(kafkaProducer, parallelism);
+ if (flushMode.isEnabled() && upsertMode) {
+ BufferedUpsertSinkFunction buffedSinkFunction =
+ new BufferedUpsertSinkFunction(
+ kafkaProducer,
+ physicalDataType,
+ keyProjection,
+ context.createTypeInformation(consumedDataType),
+ flushMode);
+ return SinkFunctionProvider.of(buffedSinkFunction, parallelism);
+ } else {
+ return SinkFunctionProvider.of(kafkaProducer, parallelism);
+ }
}
@Override
@@ -176,12 +199,14 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada
@Override
public void applyWritableMetadata(List<String> metadataKeys, DataType consumedDataType) {
this.metadataKeys = metadataKeys;
+ this.consumedDataType = consumedDataType;
}
@Override
public DynamicTableSink copy() {
final KafkaDynamicSink copy =
new KafkaDynamicSink(
+ consumedDataType,
physicalDataType,
keyEncodingFormat,
valueEncodingFormat,
@@ -193,6 +218,7 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada
partitioner,
semantic,
upsertMode,
+ flushMode,
parallelism);
copy.metadataKeys = metadataKeys;
return copy;
@@ -213,6 +239,7 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada
}
final KafkaDynamicSink that = (KafkaDynamicSink) o;
return Objects.equals(metadataKeys, that.metadataKeys)
+ && Objects.equals(consumedDataType, that.consumedDataType)
&& Objects.equals(physicalDataType, that.physicalDataType)
&& Objects.equals(keyEncodingFormat, that.keyEncodingFormat)
&& Objects.equals(valueEncodingFormat, that.valueEncodingFormat)
@@ -224,6 +251,7 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada
&& Objects.equals(partitioner, that.partitioner)
&& Objects.equals(semantic, that.semantic)
&& Objects.equals(upsertMode, that.upsertMode)
+ && Objects.equals(flushMode, that.flushMode)
&& Objects.equals(parallelism, that.parallelism);
}
@@ -231,6 +259,7 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada
public int hashCode() {
return Objects.hash(
metadataKeys,
+ consumedDataType,
physicalDataType,
keyEncodingFormat,
valueEncodingFormat,
@@ -242,6 +271,7 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada
partitioner,
semantic,
upsertMode,
+ flushMode,
parallelism);
}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.java
index 9c6f98b..cbd05cc 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.java
@@ -350,6 +350,7 @@ public class KafkaDynamicTableFactory
Integer parallelism) {
return new KafkaDynamicSink(
physicalDataType,
+ physicalDataType,
keyEncodingFormat,
valueEncodingFormat,
keyProjection,
@@ -360,6 +361,7 @@ public class KafkaDynamicTableFactory
partitioner,
semantic,
false,
+ SinkBufferFlushMode.DISABLED,
parallelism);
}
}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaOptions.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaOptions.java
index de1ffe7..993af27 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaOptions.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaOptions.java
@@ -214,6 +214,32 @@ public class KafkaOptions {
.withDescription(
"Optional semantic when commit. Valid enumerationns are [\"at-least-once\", \"exactly-once\", \"none\"]");
+ // Disable this feature by default
+ public static final ConfigOption<Integer> SINK_BUFFER_FLUSH_MAX_ROWS =
+ ConfigOptions.key("sink.buffer-flush.max-rows")
+ .intType()
+ .defaultValue(0)
+ .withDescription(
+ "The max size of buffered records before flush. "
+ + "When the sink receives many updates on the same key, the buffer will retain the last record of the same key. "
+ + "This can help to reduce data shuffling and avoid possible tombstone messages to Kafka topic."
+ + "Can be set to '0' to disable it."
+ + "Note both 'sink.buffer-flush.max-rows' and 'sink.buffer-flush.interval' "
+ + "must be set to be greater than zero to enable sink buffer flushing.");
+
+ // Disable this feature by default
+ public static final ConfigOption<Duration> SINK_BUFFER_FLUSH_INTERVAL =
+ ConfigOptions.key("sink.buffer-flush.interval")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(0))
+ .withDescription(
+ "The flush interval mills, over this time, asynchronous threads will flush data. "
+ + "When the sink receives many updates on the same key, the buffer will retain the last record of the same key. "
+ + "This can help to reduce data shuffling and avoid possible tombstone messages to Kafka topic."
+ + "Can be set to '0' to disable it. "
+ + "Note both 'sink.buffer-flush.max-rows' and 'sink.buffer-flush.interval' "
+ + "must be set to be greater than zero to enable sink buffer flushing.");
+
private static final ConfigOption<String> SCHEMA_REGISTRY_SUBJECT =
ConfigOptions.key("schema-registry.subject").stringType().noDefaultValue();
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/SinkBufferFlushMode.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/SinkBufferFlushMode.java
new file mode 100644
index 0000000..d5c0c2d
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/SinkBufferFlushMode.java
@@ -0,0 +1,79 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.streaming.connectors.kafka.table;
+
+import java.util.Objects;
+
+/** Sink buffer flush configuration. */
+public class SinkBufferFlushMode {
+
+ private static final int DISABLED_BATCH_SIZE = 0;
+ private static final long DISABLED_BATCH_INTERVAL = 0L;
+
+ public static final SinkBufferFlushMode DISABLED =
+ new SinkBufferFlushMode(DISABLED_BATCH_SIZE, DISABLED_BATCH_INTERVAL);
+
+ private final int batchSize;
+ private final long batchIntervalMs;
+
+ public SinkBufferFlushMode(int batchSize, long batchIntervalMs) {
+ this.batchSize = batchSize;
+ this.batchIntervalMs = batchIntervalMs;
+
+ // validation
+ if (isEnabled()
+ && !(batchSize > DISABLED_BATCH_SIZE
+ && batchIntervalMs > DISABLED_BATCH_INTERVAL)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "batchSize and batchInterval must greater than zero if buffer flush is enabled,"
+ + " but got batchSize=%s and batchIntervalMs=%s",
+ batchSize, batchIntervalMs));
+ }
+ }
+
+ public int getBatchSize() {
+ return batchSize;
+ }
+
+ public long getBatchIntervalMs() {
+ return batchIntervalMs;
+ }
+
+ public boolean isEnabled() {
+ return !(batchSize == DISABLED_BATCH_SIZE && batchIntervalMs == DISABLED_BATCH_INTERVAL);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SinkBufferFlushMode that = (SinkBufferFlushMode) o;
+ return batchSize == that.batchSize && batchIntervalMs == that.batchIntervalMs;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(batchSize, batchIntervalMs);
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactory.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactory.java
index fb92586..1d2860b 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactory.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactory.java
@@ -43,6 +43,7 @@ import org.apache.flink.table.factories.SerializationFormatFactory;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.RowKind;
+import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
@@ -54,6 +55,8 @@ import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.KEY
import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.KEY_FIELDS_PREFIX;
import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.KEY_FORMAT;
import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.PROPS_BOOTSTRAP_SERVERS;
+import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.SINK_BUFFER_FLUSH_INTERVAL;
+import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.SINK_BUFFER_FLUSH_MAX_ROWS;
import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.TOPIC;
import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.VALUE_FIELDS_INCLUDE;
import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.VALUE_FORMAT;
@@ -89,6 +92,8 @@ public class UpsertKafkaDynamicTableFactory
options.add(KEY_FIELDS_PREFIX);
options.add(VALUE_FIELDS_INCLUDE);
options.add(FactoryUtil.SINK_PARALLELISM);
+ options.add(SINK_BUFFER_FLUSH_INTERVAL);
+ options.add(SINK_BUFFER_FLUSH_MAX_ROWS);
return options;
}
@@ -105,7 +110,7 @@ public class UpsertKafkaDynamicTableFactory
// Validate the option data type.
helper.validateExcept(KafkaOptions.PROPERTIES_PREFIX);
TableSchema schema = context.getCatalogTable().getSchema();
- validateTableOptions(tableOptions, keyDecodingFormat, valueDecodingFormat, schema);
+ validateSource(tableOptions, keyDecodingFormat, valueDecodingFormat, schema);
Tuple2<int[], int[]> keyValueProjections =
createKeyValueProjections(context.getCatalogTable());
@@ -146,7 +151,7 @@ public class UpsertKafkaDynamicTableFactory
// Validate the option data type.
helper.validateExcept(KafkaOptions.PROPERTIES_PREFIX);
TableSchema schema = context.getCatalogTable().getSchema();
- validateTableOptions(tableOptions, keyEncodingFormat, valueEncodingFormat, schema);
+ validateSink(tableOptions, keyEncodingFormat, valueEncodingFormat, schema);
Tuple2<int[], int[]> keyValueProjections =
createKeyValueProjections(context.getCatalogTable());
@@ -155,10 +160,16 @@ public class UpsertKafkaDynamicTableFactory
Integer parallelism = tableOptions.get(FactoryUtil.SINK_PARALLELISM);
+ int batchSize = tableOptions.get(SINK_BUFFER_FLUSH_MAX_ROWS);
+ Duration batchInterval = tableOptions.get(SINK_BUFFER_FLUSH_INTERVAL);
+ SinkBufferFlushMode flushMode =
+ new SinkBufferFlushMode(batchSize, batchInterval.toMillis());
+
// use {@link org.apache.kafka.clients.producer.internals.DefaultPartitioner}.
// it will use hash partition if key is set else in round-robin behaviour.
return new KafkaDynamicSink(
schema.toPhysicalRowDataType(),
+ schema.toPhysicalRowDataType(),
keyEncodingFormat,
new EncodingFormatWrapper(valueEncodingFormat),
keyValueProjections.f0,
@@ -169,6 +180,7 @@ public class UpsertKafkaDynamicTableFactory
null,
KafkaSinkSemantic.AT_LEAST_ONCE,
true,
+ flushMode,
parallelism);
}
@@ -192,11 +204,19 @@ public class UpsertKafkaDynamicTableFactory
// Validation
// --------------------------------------------------------------------------------------------
- private static void validateTableOptions(
+ private static void validateSource(
+ ReadableConfig tableOptions, Format keyFormat, Format valueFormat, TableSchema schema) {
+ validateTopic(tableOptions);
+ validateFormat(keyFormat, valueFormat, tableOptions);
+ validatePKConstraints(schema);
+ }
+
+ private static void validateSink(
ReadableConfig tableOptions, Format keyFormat, Format valueFormat, TableSchema schema) {
validateTopic(tableOptions);
validateFormat(keyFormat, valueFormat, tableOptions);
validatePKConstraints(schema);
+ validateSinkBufferFlush(tableOptions);
}
private static void validateTopic(ReadableConfig tableOptions) {
@@ -237,6 +257,24 @@ public class UpsertKafkaDynamicTableFactory
}
}
+ private static void validateSinkBufferFlush(ReadableConfig tableOptions) {
+ int flushMaxRows = tableOptions.get(SINK_BUFFER_FLUSH_MAX_ROWS);
+ long flushIntervalMs = tableOptions.get(SINK_BUFFER_FLUSH_INTERVAL).toMillis();
+ if (flushMaxRows > 0 && flushIntervalMs > 0) {
+ // flush is enabled
+ return;
+ }
+ if (flushMaxRows <= 0 && flushIntervalMs <= 0) {
+ // flush is disabled
+ return;
+ }
+ // one of them is set which is not allowed
+ throw new ValidationException(
+ String.format(
+ "'%s' and '%s' must be set to be greater than zero together to enable sink buffer flushing.",
+ SINK_BUFFER_FLUSH_MAX_ROWS.key(), SINK_BUFFER_FLUSH_INTERVAL.key()));
+ }
+
// --------------------------------------------------------------------------------------------
// Format wrapper
// --------------------------------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/BufferedUpsertSinkFunctionTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/BufferedUpsertSinkFunctionTest.java
new file mode 100644
index 0000000..e8d6a9a
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/BufferedUpsertSinkFunctionTest.java
@@ -0,0 +1,330 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.streaming.connectors.kafka.table;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.state.CheckpointListener;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
+import org.apache.flink.streaming.util.MockStreamingRuntimeContext;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext;
+
+import org.junit.Test;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.types.RowKind.DELETE;
+import static org.apache.flink.types.RowKind.INSERT;
+import static org.apache.flink.types.RowKind.UPDATE_AFTER;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/** Test for {@link BufferedUpsertSinkFunction}. */
+public class BufferedUpsertSinkFunctionTest {
+
+ private static final ResolvedSchema SCHEMA =
+ ResolvedSchema.of(
+ Column.physical("id", DataTypes.INT().notNull()),
+ Column.physical("title", DataTypes.STRING().notNull()),
+ Column.physical("author", DataTypes.STRING()),
+ Column.physical("price", DataTypes.DOUBLE()),
+ Column.physical("qty", DataTypes.INT()),
+ Column.physical("ts", DataTypes.TIMESTAMP_LTZ(3)));
+
+ private static final int keyIndices = 0;
+ private static final int TIMESTAMP_INDICES = 5;
+ private static final SinkBufferFlushMode BUFFER_FLUSH_MODE =
+ new SinkBufferFlushMode(4, Long.MAX_VALUE);
+
+ public static final RowData[] TEST_DATA = {
+ GenericRowData.ofKind(
+ INSERT,
+ 1001,
+ StringData.fromString("Java public for dummies"),
+ StringData.fromString("Tan Ah Teck"),
+ 11.11,
+ 11,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T15:00:00Z"))),
+ GenericRowData.ofKind(
+ INSERT,
+ 1002,
+ StringData.fromString("More Java for dummies"),
+ StringData.fromString("Tan Ah Teck"),
+ 22.22,
+ 22,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T16:00:00Z"))),
+ GenericRowData.ofKind(
+ INSERT,
+ 1004,
+ StringData.fromString("A Cup of Java"),
+ StringData.fromString("Kumar"),
+ 44.44,
+ 44,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T17:00:00Z"))),
+ GenericRowData.ofKind(
+ UPDATE_AFTER,
+ 1004,
+ StringData.fromString("A Teaspoon of Java"),
+ StringData.fromString("Kevin Jones"),
+ 55.55,
+ 55,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T18:00:00Z"))),
+ GenericRowData.ofKind(
+ UPDATE_AFTER,
+ 1004,
+ StringData.fromString("A Teaspoon of Java 1.4"),
+ StringData.fromString("Kevin Jones"),
+ 66.66,
+ 66,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T19:00:00Z"))),
+ GenericRowData.ofKind(
+ UPDATE_AFTER,
+ 1004,
+ StringData.fromString("A Teaspoon of Java 1.5"),
+ StringData.fromString("Kevin Jones"),
+ 77.77,
+ 77,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T20:00:00Z"))),
+ GenericRowData.ofKind(
+ DELETE,
+ 1004,
+ StringData.fromString("A Teaspoon of Java 1.8"),
+ StringData.fromString("Kevin Jones"),
+ null,
+ 1010,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T21:00:00Z")))
+ };
+
+ @Test
+ public void testWriteData() throws Exception {
+ MockedSinkFunction sinkFunction = new MockedSinkFunction();
+ BufferedUpsertSinkFunction bufferedSink = createBufferedSink(sinkFunction);
+
+ // write 3 records which doesn't trigger batch size
+ writeData(bufferedSink, TEST_DATA, 0, 3);
+ assertTrue(sinkFunction.rowDataCollectors.isEmpty());
+
+ // write one more record, and should flush the buffer
+ writeData(bufferedSink, TEST_DATA, 3, 1);
+
+ HashMap<Integer, List<RowData>> expected = new HashMap<>();
+ expected.put(
+ 1001,
+ Collections.singletonList(
+ GenericRowData.ofKind(
+ UPDATE_AFTER,
+ 1001,
+ StringData.fromString("Java public for dummies"),
+ StringData.fromString("Tan Ah Teck"),
+ 11.11,
+ 11,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T15:00:00Z")))));
+ expected.put(
+ 1002,
+ Collections.singletonList(
+ GenericRowData.ofKind(
+ UPDATE_AFTER,
+ 1002,
+ StringData.fromString("More Java for dummies"),
+ StringData.fromString("Tan Ah Teck"),
+ 22.22,
+ 22,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T16:00:00Z")))));
+ expected.put(
+ 1004,
+ Collections.singletonList(
+ GenericRowData.ofKind(
+ UPDATE_AFTER,
+ 1004,
+ StringData.fromString("A Teaspoon of Java"),
+ StringData.fromString("Kevin Jones"),
+ 55.55,
+ 55,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T18:00:00Z")))));
+
+ compareCompactedResult(expected, sinkFunction.rowDataCollectors);
+
+ sinkFunction.rowDataCollectors.clear();
+ // write remaining data, and they are still buffered
+ writeData(bufferedSink, TEST_DATA, 4, 3);
+ assertTrue(sinkFunction.rowDataCollectors.isEmpty());
+ }
+
+ @Test
+ public void testFlushDataWhenCheckpointing() throws Exception {
+ MockedSinkFunction sinkFunction = new MockedSinkFunction();
+ BufferedUpsertSinkFunction bufferedFunction = createBufferedSink(sinkFunction);
+ // write all data, there should be 3 records are still buffered
+ writeData(bufferedFunction, TEST_DATA, 0, TEST_DATA.length);
+ // snapshot should flush the buffer
+ bufferedFunction.snapshotState(null);
+
+ HashMap<Integer, List<RowData>> expected = new HashMap<>();
+ expected.put(
+ 1001,
+ Collections.singletonList(
+ GenericRowData.ofKind(
+ UPDATE_AFTER,
+ 1001,
+ StringData.fromString("Java public for dummies"),
+ StringData.fromString("Tan Ah Teck"),
+ 11.11,
+ 11,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T15:00:00Z")))));
+ expected.put(
+ 1002,
+ Collections.singletonList(
+ GenericRowData.ofKind(
+ UPDATE_AFTER,
+ 1002,
+ StringData.fromString("More Java for dummies"),
+ StringData.fromString("Tan Ah Teck"),
+ 22.22,
+ 22,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T16:00:00Z")))));
+ expected.put(
+ 1004,
+ Arrays.asList(
+ GenericRowData.ofKind(
+ UPDATE_AFTER,
+ 1004,
+ StringData.fromString("A Teaspoon of Java"),
+ StringData.fromString("Kevin Jones"),
+ 55.55,
+ 55,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T18:00:00Z"))),
+ GenericRowData.ofKind(
+ DELETE,
+ 1004,
+ StringData.fromString("A Teaspoon of Java 1.8"),
+ StringData.fromString("Kevin Jones"),
+ null,
+ 1010,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T21:00:00Z")))));
+
+ compareCompactedResult(expected, sinkFunction.rowDataCollectors);
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @SuppressWarnings("unchecked")
+ private BufferedUpsertSinkFunction createBufferedSink(MockedSinkFunction sinkFunction)
+ throws Exception {
+ TypeInformation<RowData> typeInformation =
+ (TypeInformation<RowData>)
+ new SinkRuntimeProviderContext(false)
+ .createTypeInformation(SCHEMA.toPhysicalRowDataType());
+
+ BufferedUpsertSinkFunction bufferedSinkFunction =
+ new BufferedUpsertSinkFunction(
+ sinkFunction,
+ SCHEMA.toPhysicalRowDataType(),
+ new int[] {keyIndices},
+ typeInformation,
+ BUFFER_FLUSH_MODE);
+ bufferedSinkFunction.open(new Configuration());
+
+ return bufferedSinkFunction;
+ }
+
+ private void writeData(BufferedUpsertSinkFunction sink, RowData[] data, int startPos, int size)
+ throws Exception {
+ for (int i = startPos; i < startPos + size; i++) {
+ RowData row = data[i];
+ long rowtime = row.getTimestamp(TIMESTAMP_INDICES, 3).getMillisecond();
+ sink.invoke(row, SinkContextUtil.forTimestamp(rowtime));
+ }
+ }
+
+ private void compareCompactedResult(
+ Map<Integer, List<RowData>> expected, List<RowData> actual) {
+ Map<Integer, List<RowData>> actualMap = new HashMap<>();
+
+ for (RowData rowData : actual) {
+ Integer id = rowData.getInt(keyIndices);
+ actualMap.computeIfAbsent(id, key -> new ArrayList<>()).add(rowData);
+ }
+
+ assertEquals(expected.size(), actualMap.size());
+ for (Integer id : expected.keySet()) {
+ assertEquals(expected.get(id), actualMap.get(id));
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private static class MockedSinkFunction extends RichSinkFunction<RowData>
+ implements CheckpointedFunction, CheckpointListener {
+
+ private static final long serialVersionUID = 1L;
+ transient List<RowData> rowDataCollectors;
+
+ @Override
+ public RuntimeContext getRuntimeContext() {
+ return new MockStreamingRuntimeContext(true, 1, 1);
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ rowDataCollectors = new ArrayList<>();
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {
+ // do nothing
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) {
+ // do nothing
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) {
+ // do nothing
+ }
+
+ @Override
+ public void invoke(RowData value, Context context) {
+ assertEquals(
+ value.getTimestamp(TIMESTAMP_INDICES, 3).toInstant(),
+ Instant.ofEpochMilli(context.timestamp()));
+ rowDataCollectors.add(value);
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java
index 027391d..15e2913 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java
@@ -886,6 +886,7 @@ public class KafkaDynamicTableFactoryTest extends TestLogger {
@Nullable Integer parallelism) {
return new KafkaDynamicSink(
physicalDataType,
+ physicalDataType,
keyEncodingFormat,
valueEncodingFormat,
keyProjection,
@@ -896,6 +897,7 @@ public class KafkaDynamicTableFactoryTest extends TestLogger {
partitioner,
semantic,
false,
+ SinkBufferFlushMode.DISABLED,
parallelism);
}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactoryTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactoryTest.java
index d05845e..07e155d 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactoryTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactoryTest.java
@@ -119,20 +119,23 @@ public class UpsertKafkaDynamicTableFactoryTest extends TestLogger {
UPSERT_KAFKA_SINK_PROPERTIES.setProperty("bootstrap.servers", "dummy");
}
+ static EncodingFormat<SerializationSchema<RowData>> keyEncodingFormat =
+ new TestFormatFactory.EncodingFormatMock(",", ChangelogMode.insertOnly());
+ static EncodingFormat<SerializationSchema<RowData>> valueEncodingFormat =
+ new TestFormatFactory.EncodingFormatMock(",", ChangelogMode.insertOnly());
+
+ static DecodingFormat<DeserializationSchema<RowData>> keyDecodingFormat =
+ new TestFormatFactory.DecodingFormatMock(
+ ",", true, ChangelogMode.insertOnly(), Collections.emptyMap());
+ static DecodingFormat<DeserializationSchema<RowData>> valueDecodingFormat =
+ new TestFormatFactory.DecodingFormatMock(
+ ",", true, ChangelogMode.insertOnly(), Collections.emptyMap());
+
@Rule public ExpectedException thrown = ExpectedException.none();
@Test
public void testTableSource() {
final DataType producedDataType = SOURCE_SCHEMA.toPhysicalRowDataType();
-
- DecodingFormat<DeserializationSchema<RowData>> keyDecodingFormat =
- new TestFormatFactory.DecodingFormatMock(
- ",", true, ChangelogMode.insertOnly(), Collections.emptyMap());
-
- DecodingFormat<DeserializationSchema<RowData>> valueDecodingFormat =
- new TestFormatFactory.DecodingFormatMock(
- ",", true, ChangelogMode.insertOnly(), Collections.emptyMap());
-
// Construct table source using options and table source factory
final DynamicTableSource actualSource =
createTableSource(SOURCE_SCHEMA, getFullSourceOptions());
@@ -161,11 +164,6 @@ public class UpsertKafkaDynamicTableFactoryTest extends TestLogger {
@Test
public void testTableSink() {
- EncodingFormat<SerializationSchema<RowData>> keyEncodingFormat =
- new TestFormatFactory.EncodingFormatMock(",", ChangelogMode.insertOnly());
- EncodingFormat<SerializationSchema<RowData>> valueEncodingFormat =
- new TestFormatFactory.EncodingFormatMock(",", ChangelogMode.insertOnly());
-
// Construct table sink using options and table sink factory.
final DynamicTableSink actualSink = createTableSink(SINK_SCHEMA, getFullSinkOptions());
@@ -179,6 +177,7 @@ public class UpsertKafkaDynamicTableFactoryTest extends TestLogger {
null,
SINK_TOPIC,
UPSERT_KAFKA_SINK_PROPERTIES,
+ SinkBufferFlushMode.DISABLED,
null);
// Test sink format.
@@ -195,17 +194,51 @@ public class UpsertKafkaDynamicTableFactoryTest extends TestLogger {
}
@Test
+ public void testBufferedTableSink() {
+ // Construct table sink using options and table sink factory.
+ final DynamicTableSink actualSink =
+ createTableSink(
+ SINK_SCHEMA,
+ getModifiedOptions(
+ getFullSinkOptions(),
+ options -> {
+ options.put("sink.buffer-flush.max-rows", "100");
+ options.put("sink.buffer-flush.interval", "1s");
+ }));
+
+ final DynamicTableSink expectedSink =
+ createExpectedSink(
+ SINK_SCHEMA.toPhysicalRowDataType(),
+ keyEncodingFormat,
+ valueEncodingFormat,
+ SINK_KEY_FIELDS,
+ SINK_VALUE_FIELDS,
+ null,
+ SINK_TOPIC,
+ UPSERT_KAFKA_SINK_PROPERTIES,
+ new SinkBufferFlushMode(100, 1000L),
+ null);
+
+ // Test sink format.
+ final KafkaDynamicSink actualUpsertKafkaSink = (KafkaDynamicSink) actualSink;
+ assertEquals(expectedSink, actualSink);
+
+ // Test kafka producer.
+ DynamicTableSink.SinkRuntimeProvider provider =
+ actualUpsertKafkaSink.getSinkRuntimeProvider(new SinkRuntimeProviderContext(false));
+ assertThat(provider, instanceOf(SinkFunctionProvider.class));
+ final SinkFunctionProvider sinkFunctionProvider = (SinkFunctionProvider) provider;
+ final SinkFunction<RowData> sinkFunction = sinkFunctionProvider.createSinkFunction();
+ assertThat(sinkFunction, instanceOf(BufferedUpsertSinkFunction.class));
+ }
+
+ @Test
public void testTableSinkWithParallelism() {
final Map<String, String> modifiedOptions =
getModifiedOptions(
getFullSinkOptions(), options -> options.put("sink.parallelism", "100"));
final DynamicTableSink actualSink = createTableSink(SINK_SCHEMA, modifiedOptions);
- EncodingFormat<SerializationSchema<RowData>> keyEncodingFormat =
- new TestFormatFactory.EncodingFormatMock(",", ChangelogMode.insertOnly());
- EncodingFormat<SerializationSchema<RowData>> valueEncodingFormat =
- new TestFormatFactory.EncodingFormatMock(",", ChangelogMode.insertOnly());
-
final DynamicTableSink expectedSink =
createExpectedSink(
SINK_SCHEMA.toPhysicalRowDataType(),
@@ -216,6 +249,7 @@ public class UpsertKafkaDynamicTableFactoryTest extends TestLogger {
null,
SINK_TOPIC,
UPSERT_KAFKA_SINK_PROPERTIES,
+ SinkBufferFlushMode.DISABLED,
100);
assertEquals(expectedSink, actualSink);
@@ -422,6 +456,25 @@ public class UpsertKafkaDynamicTableFactoryTest extends TestLogger {
"I;UA;UB;D")));
}
+ @Test
+ public void testInvalidSinkBufferFlush() {
+ thrown.expect(ValidationException.class);
+ thrown.expect(
+ containsCause(
+ new ValidationException(
+ "'sink.buffer-flush.max-rows' and 'sink.buffer-flush.interval' "
+ + "must be set to be greater than zero together to enable"
+ + " sink buffer flushing.")));
+ createTableSink(
+ SINK_SCHEMA,
+ getModifiedOptions(
+ getFullSinkOptions(),
+ options -> {
+ options.put("sink.buffer-flush.max-rows", "0");
+ options.put("sink.buffer-flush.interval", "1s");
+ }));
+ }
+
// --------------------------------------------------------------------------------------------
// Utilities
// --------------------------------------------------------------------------------------------
@@ -546,9 +599,11 @@ public class UpsertKafkaDynamicTableFactoryTest extends TestLogger {
String keyPrefix,
String topic,
Properties properties,
+ SinkBufferFlushMode flushMode,
Integer parallelism) {
return new KafkaDynamicSink(
consumedDataType,
+ consumedDataType,
keyEncodingFormat,
new UpsertKafkaDynamicTableFactory.EncodingFormatWrapper(valueEncodingFormat),
keyProjection,
@@ -559,6 +614,7 @@ public class UpsertKafkaDynamicTableFactoryTest extends TestLogger {
null,
KafkaSinkSemantic.AT_LEAST_ONCE,
true,
+ flushMode,
parallelism);
}
}