You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/08/14 07:31:39 UTC
[flink] branch master updated (adee561 -> c082a10)
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from adee561 [FLINK-19554][connector/testing-framework] KafkaSource IT and E2E case based on connector testing framework
new 134f388 [FLINK-23621] Add InterruptedException to ProcessingTimeService#onProcessingTime
new c082a10 [FLINK-23735][connectors/kafka] Migrate BufferedUpsertSinkFunction to FLIP-143 Sink API
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../connector/file/sink/writer/FileWriterTest.java | 6 +-
.../kafka/table/BufferedUpsertSinkFunction.java | 316 ---------------------
.../table/DynamicKafkaSerializationSchema.java | 198 -------------
.../connectors/kafka/table/KafkaDynamicSink.java | 135 +++++----
.../connectors/kafka/table/ReducingUpsertSink.java | 99 +++++++
.../kafka/table/ReducingUpsertWriter.java | 183 ++++++++++++
...tionTest.java => ReducingUpsertWriterTest.java} | 216 +++++++-------
.../table/UpsertKafkaDynamicTableFactoryTest.java | 29 +-
.../org/apache/flink/api/connector/sink/Sink.java | 2 +-
.../operators/sink/SinkOperatorFactory.java | 6 +
10 files changed, 490 insertions(+), 700 deletions(-)
delete mode 100644 flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/BufferedUpsertSinkFunction.java
delete mode 100644 flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaSerializationSchema.java
create mode 100644 flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertSink.java
create mode 100644 flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriter.java
rename flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/{BufferedUpsertSinkFunctionTest.java => ReducingUpsertWriterTest.java} (72%)
[flink] 02/02: [FLINK-23735][connectors/kafka] Migrate
BufferedUpsertSinkFunction to FLIP-143 Sink API
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c082a10d5d3087961184e520476363849e8b3a8a
Author: Fabian Paul <fa...@ververica.com>
AuthorDate: Thu Aug 12 13:57:12 2021 +0200
[FLINK-23735][connectors/kafka] Migrate BufferedUpsertSinkFunction to FLIP-143 Sink API
---
.../kafka/table/BufferedUpsertSinkFunction.java | 316 ---------------------
.../table/DynamicKafkaSerializationSchema.java | 198 -------------
.../connectors/kafka/table/KafkaDynamicSink.java | 135 +++++----
.../connectors/kafka/table/ReducingUpsertSink.java | 99 +++++++
.../kafka/table/ReducingUpsertWriter.java | 183 ++++++++++++
...tionTest.java => ReducingUpsertWriterTest.java} | 216 +++++++-------
.../table/UpsertKafkaDynamicTableFactoryTest.java | 29 +-
.../operators/sink/SinkOperatorFactory.java | 6 +
8 files changed, 486 insertions(+), 696 deletions(-)
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
deleted file mode 100644
index 9a267b0..0000000
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/BufferedUpsertSinkFunction.java
+++ /dev/null
@@ -1,316 +0,0 @@
-/*
- * 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.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.apache.flink.util.concurrent.ExecutorThreadFactory;
-
-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.DynamicKafkaRecordSerializationSchema.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()
- ? typeSerializer::copy
- : Function.identity();
-
- // 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 void finish() {
- 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);
- }
- }
- }
-
- @Override
- public synchronized void close() throws Exception {
- if (!closed) {
- closed = true;
-
- if (this.scheduledFuture != null) {
- scheduledFuture.cancel(false);
- this.scheduler.shutdown();
- }
-
- 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
deleted file mode 100644
index 590ea51..0000000
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaSerializationSchema.java
+++ /dev/null
@@ -1,198 +0,0 @@
-/*
- * 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.serialization.SerializationSchema;
-import org.apache.flink.streaming.connectors.kafka.KafkaContextAware;
-import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
-import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.types.RowKind;
-import org.apache.flink.util.Preconditions;
-
-import org.apache.kafka.clients.producer.ProducerRecord;
-
-import javax.annotation.Nullable;
-
-import java.io.Serializable;
-
-/** A specific {@link KafkaSerializationSchema} for {@link KafkaDynamicSink}. */
-class DynamicKafkaSerializationSchema
- implements KafkaSerializationSchema<RowData>, KafkaContextAware<RowData> {
-
- private static final long serialVersionUID = 1L;
-
- private final @Nullable FlinkKafkaPartitioner<RowData> partitioner;
-
- private final String topic;
-
- private final @Nullable SerializationSchema<RowData> keySerialization;
-
- private final SerializationSchema<RowData> valueSerialization;
-
- private final RowData.FieldGetter[] keyFieldGetters;
-
- private final RowData.FieldGetter[] valueFieldGetters;
-
- private final boolean hasMetadata;
-
- private final boolean upsertMode;
-
- /**
- * Contains the position for each value of {@link KafkaDynamicSink.WritableMetadata} in the
- * consumed row or -1 if this metadata key is not used.
- */
- private final int[] metadataPositions;
-
- private int[] partitions;
-
- private int parallelInstanceId;
-
- private int numParallelInstances;
-
- DynamicKafkaSerializationSchema(
- String topic,
- @Nullable FlinkKafkaPartitioner<RowData> partitioner,
- @Nullable SerializationSchema<RowData> keySerialization,
- SerializationSchema<RowData> valueSerialization,
- RowData.FieldGetter[] keyFieldGetters,
- RowData.FieldGetter[] valueFieldGetters,
- boolean hasMetadata,
- int[] metadataPositions,
- boolean upsertMode) {
- if (upsertMode) {
- Preconditions.checkArgument(
- keySerialization != null && keyFieldGetters.length > 0,
- "Key must be set in upsert mode for serialization schema.");
- }
- this.topic = topic;
- this.partitioner = partitioner;
- this.keySerialization = keySerialization;
- this.valueSerialization = valueSerialization;
- this.keyFieldGetters = keyFieldGetters;
- this.valueFieldGetters = valueFieldGetters;
- this.hasMetadata = hasMetadata;
- this.metadataPositions = metadataPositions;
- this.upsertMode = upsertMode;
- }
-
- @Override
- public void open(SerializationSchema.InitializationContext context) throws Exception {
- if (keySerialization != null) {
- keySerialization.open(context);
- }
- valueSerialization.open(context);
- if (partitioner != null) {
- partitioner.open(parallelInstanceId, numParallelInstances);
- }
- }
-
- @Override
- public ProducerRecord<byte[], byte[]> serialize(RowData consumedRow, @Nullable Long timestamp) {
- // shortcut in case no input projection is required
- if (keySerialization == null && !hasMetadata) {
- final byte[] valueSerialized = valueSerialization.serialize(consumedRow);
- return new ProducerRecord<>(
- topic,
- extractPartition(consumedRow, null, valueSerialized),
- null,
- valueSerialized);
- }
-
- final byte[] keySerialized;
- if (keySerialization == null) {
- keySerialized = null;
- } else {
- final RowData keyRow =
- DynamicKafkaRecordSerializationSchema.createProjectedRow(
- consumedRow, RowKind.INSERT, keyFieldGetters);
- keySerialized = keySerialization.serialize(keyRow);
- }
-
- final byte[] valueSerialized;
- final RowKind kind = consumedRow.getRowKind();
- final RowData valueRow =
- DynamicKafkaRecordSerializationSchema.createProjectedRow(
- consumedRow, kind, valueFieldGetters);
- if (upsertMode) {
- if (kind == RowKind.DELETE || kind == RowKind.UPDATE_BEFORE) {
- // transform the message as the tombstone message
- valueSerialized = null;
- } else {
- // make the message to be INSERT to be compliant with the INSERT-ONLY format
- valueRow.setRowKind(RowKind.INSERT);
- valueSerialized = valueSerialization.serialize(valueRow);
- }
- } else {
- valueSerialized = valueSerialization.serialize(valueRow);
- }
-
- return new ProducerRecord<>(
- topic,
- extractPartition(consumedRow, keySerialized, valueSerialized),
- readMetadata(consumedRow, KafkaDynamicSink.WritableMetadata.TIMESTAMP),
- keySerialized,
- valueSerialized,
- readMetadata(consumedRow, KafkaDynamicSink.WritableMetadata.HEADERS));
- }
-
- @Override
- public void setParallelInstanceId(int parallelInstanceId) {
- this.parallelInstanceId = parallelInstanceId;
- }
-
- @Override
- public void setNumParallelInstances(int numParallelInstances) {
- this.numParallelInstances = numParallelInstances;
- }
-
- @Override
- public void setPartitions(int[] partitions) {
- this.partitions = partitions;
- }
-
- @Override
- public String getTargetTopic(RowData element) {
- return topic;
- }
-
- @SuppressWarnings("unchecked")
- private <T> T readMetadata(RowData consumedRow, KafkaDynamicSink.WritableMetadata metadata) {
- final int pos = metadataPositions[metadata.ordinal()];
- if (pos < 0) {
- return null;
- }
- return (T) metadata.converter.read(consumedRow, pos);
- }
-
- private Integer extractPartition(
- RowData consumedRow, @Nullable byte[] keySerialized, byte[] valueSerialized) {
- if (partitioner != null) {
- return partitioner.partition(
- consumedRow, keySerialized, valueSerialized, topic, partitions);
- }
- return null;
- }
-
- // --------------------------------------------------------------------------------------------
-
- interface MetadataConverter extends Serializable {
- Object read(RowData consumedRow, int pos);
- }
-}
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 cfd93f2..9fd5af8 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
@@ -19,18 +19,21 @@
package org.apache.flink.streaming.connectors.kafka.table;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.connectors.kafka.sink.KafkaSink;
import org.apache.flink.streaming.connectors.kafka.sink.KafkaSinkBuilder;
-import org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaSerializationSchema.MetadataConverter;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
import org.apache.flink.table.connector.sink.DynamicTableSink;
-import org.apache.flink.table.connector.sink.SinkFunctionProvider;
import org.apache.flink.table.connector.sink.SinkProvider;
import org.apache.flink.table.connector.sink.abilities.SupportsWritingMetadata;
import org.apache.flink.table.data.ArrayData;
@@ -45,6 +48,7 @@ import org.apache.kafka.common.header.Header;
import javax.annotation.Nullable;
+import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -53,6 +57,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
+import java.util.function.Function;
import java.util.stream.Stream;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -185,44 +190,57 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada
final SerializationSchema<RowData> valueSerialization =
createSerialization(context, valueEncodingFormat, valueProjection, null);
+ final KafkaSinkBuilder<RowData> sinkBuilder = KafkaSink.builder();
+ final List<LogicalType> physicalChildren = physicalDataType.getLogicalType().getChildren();
+ if (transactionalIdPrefix != null) {
+ sinkBuilder.setTransactionalIdPrefix(transactionalIdPrefix);
+ }
+ final KafkaSink<RowData> kafkaSink =
+ sinkBuilder
+ .setDeliverGuarantee(deliveryGuarantee)
+ .setBootstrapServers(
+ properties.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG).toString())
+ .setKafkaProducerConfig(properties)
+ .setRecordSerializer(
+ new DynamicKafkaRecordSerializationSchema(
+ topic,
+ partitioner,
+ keySerialization,
+ valueSerialization,
+ getFieldGetters(physicalChildren, keyProjection),
+ getFieldGetters(physicalChildren, valueProjection),
+ hasMetadata(),
+ getMetadataPositions(physicalChildren),
+ upsertMode))
+ .build();
if (flushMode.isEnabled() && upsertMode) {
- BufferedUpsertSinkFunction buffedSinkFunction =
- new BufferedUpsertSinkFunction(
- createKafkaProducer(keySerialization, valueSerialization),
- physicalDataType,
- keyProjection,
- context.createTypeInformation(consumedDataType),
- flushMode);
- return SinkFunctionProvider.of(buffedSinkFunction, parallelism);
- } else {
- final KafkaSinkBuilder<RowData> sinkBuilder = KafkaSink.builder();
- final List<LogicalType> physicalChildren =
- physicalDataType.getLogicalType().getChildren();
- if (transactionalIdPrefix != null) {
- sinkBuilder.setTransactionalIdPrefix(transactionalIdPrefix);
- }
- return SinkProvider.of(
- sinkBuilder
- .setDeliverGuarantee(deliveryGuarantee)
- .setBootstrapServers(
- properties
- .get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG)
- .toString())
- .setKafkaProducerConfig(properties)
- .setRecordSerializer(
- new DynamicKafkaRecordSerializationSchema(
- topic,
- partitioner,
- keySerialization,
- valueSerialization,
- getFieldGetters(physicalChildren, keyProjection),
- getFieldGetters(physicalChildren, valueProjection),
- hasMetadata(),
- getMetadataPositions(physicalChildren),
- upsertMode))
- .build(),
- parallelism);
+ return (DataStreamSinkProvider)
+ dataStream -> {
+ final boolean objectReuse =
+ dataStream
+ .getExecutionEnvironment()
+ .getConfig()
+ .isObjectReuseEnabled();
+ final ReducingUpsertSink<?> sink =
+ new ReducingUpsertSink<>(
+ kafkaSink,
+ physicalDataType,
+ keyProjection,
+ flushMode,
+ objectReuse
+ ? createRowDataTypeSerializer(
+ context,
+ dataStream.getExecutionConfig())
+ ::copy
+ : Function.identity());
+ final DataStreamSink<RowData> end = dataStream.sinkTo(sink);
+ if (parallelism != null) {
+ end.setParallelism(parallelism);
+ }
+ return end;
+ };
}
+ return SinkProvider.of(kafkaSink, parallelism);
}
@Override
@@ -317,38 +335,11 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada
// --------------------------------------------------------------------------------------------
- protected FlinkKafkaProducer<RowData> createKafkaProducer(
- SerializationSchema<RowData> keySerialization,
- SerializationSchema<RowData> valueSerialization) {
- final List<LogicalType> physicalChildren = physicalDataType.getLogicalType().getChildren();
-
- final RowData.FieldGetter[] keyFieldGetters =
- getFieldGetters(physicalChildren, keyProjection);
-
- final RowData.FieldGetter[] valueFieldGetters =
- getFieldGetters(physicalChildren, valueProjection);
-
- // determine the positions of metadata in the consumed row
- final int[] metadataPositions = getMetadataPositions(physicalChildren);
-
- final DynamicKafkaSerializationSchema kafkaSerializer =
- new DynamicKafkaSerializationSchema(
- topic,
- partitioner,
- keySerialization,
- valueSerialization,
- keyFieldGetters,
- valueFieldGetters,
- hasMetadata(),
- metadataPositions,
- upsertMode);
-
- return new FlinkKafkaProducer<>(
- topic,
- kafkaSerializer,
- properties,
- getSemantic(deliveryGuarantee),
- FlinkKafkaProducer.DEFAULT_KAFKA_PRODUCERS_POOL_SIZE);
+ private TypeSerializer<RowData> createRowDataTypeSerializer(
+ Context context, ExecutionConfig executionConfig) {
+ final TypeInformation<RowData> typeInformation =
+ context.createTypeInformation(consumedDataType);
+ return typeInformation.createSerializer(executionConfig);
}
private int[] getMetadataPositions(List<LogicalType> physicalChildren) {
@@ -469,6 +460,10 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada
}
}
+ interface MetadataConverter extends Serializable {
+ Object read(RowData consumedRow, int pos);
+ }
+
// --------------------------------------------------------------------------------------------
private static class KafkaHeader implements Header {
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertSink.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertSink.java
new file mode 100644
index 0000000..1cb7231
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertSink.java
@@ -0,0 +1,99 @@
+/*
+ * 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.connector.sink.Committer;
+import org.apache.flink.api.connector.sink.GlobalCommitter;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.api.connector.sink.SinkWriter;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Function;
+
+/**
+ * A wrapper of a {@link Sink}. It will buffer the data emitted by the wrapper {@link SinkWriter}
+ * and only emit it when the buffer is full or a timer is triggered or a checkpoint happens.
+ *
+ * <p>The sink provides eventual consistency guarantees without the need of a two-phase protocol
+ * because the updates are idempotent therefore duplicates have no effect.
+ */
+class ReducingUpsertSink<WriterState> implements Sink<RowData, Void, WriterState, Void> {
+
+ private final Sink<RowData, ?, WriterState, ?> wrappedSink;
+ private final DataType physicalDataType;
+ private final int[] keyProjection;
+ private final SinkBufferFlushMode bufferFlushMode;
+ private final Function<RowData, RowData> valueCopyFunction;
+
+ ReducingUpsertSink(
+ Sink<RowData, ?, WriterState, ?> wrappedSink,
+ DataType physicalDataType,
+ int[] keyProjection,
+ SinkBufferFlushMode bufferFlushMode,
+ Function<RowData, RowData> valueCopyFunction) {
+ this.wrappedSink = wrappedSink;
+ this.physicalDataType = physicalDataType;
+ this.keyProjection = keyProjection;
+ this.bufferFlushMode = bufferFlushMode;
+ this.valueCopyFunction = valueCopyFunction;
+ }
+
+ @Override
+ public SinkWriter<RowData, Void, WriterState> createWriter(
+ InitContext context, List<WriterState> states) throws IOException {
+ final SinkWriter<RowData, ?, WriterState> wrapperWriter =
+ wrappedSink.createWriter(context, states);
+ return new ReducingUpsertWriter<>(
+ wrapperWriter,
+ physicalDataType,
+ keyProjection,
+ bufferFlushMode,
+ context.getProcessingTimeService(),
+ valueCopyFunction);
+ }
+
+ @Override
+ public Optional<SimpleVersionedSerializer<WriterState>> getWriterStateSerializer() {
+ return wrappedSink.getWriterStateSerializer();
+ }
+
+ @Override
+ public Optional<Committer<Void>> createCommitter() throws IOException {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalCommitter<Void, Void>> createGlobalCommitter() throws IOException {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<SimpleVersionedSerializer<Void>> getCommittableSerializer() {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<SimpleVersionedSerializer<Void>> getGlobalCommittableSerializer() {
+ return Optional.empty();
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriter.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriter.java
new file mode 100644
index 0000000..aa178af
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriter.java
@@ -0,0 +1,183 @@
+/*
+ * 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.connector.sink.Sink;
+import org.apache.flink.api.connector.sink.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple2;
+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 java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+
+import static org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaRecordSerializationSchema.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;
+
+class ReducingUpsertWriter<WriterState> implements SinkWriter<RowData, Void, WriterState> {
+
+ private final SinkWriter<RowData, ?, WriterState> wrappedWriter;
+ private final WrappedContext wrappedContext = new WrappedContext();
+ private final int batchMaxRowNums;
+ private final Function<RowData, RowData> valueCopyFunction;
+ private final Map<RowData, Tuple2<RowData, Long>> reduceBuffer = new HashMap<>();
+ private final Function<RowData, RowData> keyExtractor;
+ private final Sink.ProcessingTimeService timeService;
+ private final long batchIntervalMs;
+
+ private boolean closed = false;
+ private long lastFlush = System.currentTimeMillis();
+
+ ReducingUpsertWriter(
+ SinkWriter<RowData, ?, WriterState> wrappedWriter,
+ DataType physicalDataType,
+ int[] keyProjection,
+ SinkBufferFlushMode bufferFlushMode,
+ Sink.ProcessingTimeService timeService,
+ Function<RowData, RowData> valueCopyFunction) {
+ checkArgument(bufferFlushMode != null && bufferFlushMode.isEnabled());
+ this.wrappedWriter = checkNotNull(wrappedWriter);
+ this.timeService = checkNotNull(timeService);
+ this.batchMaxRowNums = bufferFlushMode.getBatchSize();
+ this.batchIntervalMs = bufferFlushMode.getBatchIntervalMs();
+ registerFlush();
+ 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);
+ this.valueCopyFunction = valueCopyFunction;
+ }
+
+ @Override
+ public void write(RowData element, Context context) throws IOException, InterruptedException {
+ wrappedContext.setContext(context);
+ addToBuffer(element, context.timestamp());
+ }
+
+ @Override
+ public List<Void> prepareCommit(boolean flush) throws IOException, InterruptedException {
+ flush();
+ return Collections.emptyList();
+ }
+
+ @Override
+ public List<WriterState> snapshotState() throws IOException {
+ return wrappedWriter.snapshotState();
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (!closed) {
+ closed = true;
+ wrappedWriter.close();
+ }
+ }
+
+ private void addToBuffer(RowData row, Long timestamp) throws IOException, InterruptedException {
+ RowData key = keyExtractor.apply(row);
+ RowData value = valueCopyFunction.apply(row);
+ reduceBuffer.put(key, new Tuple2<>(changeFlag(value), timestamp));
+
+ if (reduceBuffer.size() >= batchMaxRowNums) {
+ flush();
+ }
+ }
+
+ private void registerFlush() {
+ if (closed) {
+ return;
+ }
+ timeService.registerProcessingTimer(
+ lastFlush + batchIntervalMs,
+ (t) -> {
+ if (t >= lastFlush + batchIntervalMs) {
+ flush();
+ }
+ registerFlush();
+ });
+ }
+
+ 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 flush() throws IOException, InterruptedException {
+ for (Tuple2<RowData, Long> value : reduceBuffer.values()) {
+ wrappedContext.setTimestamp(value.f1);
+ wrappedWriter.write(value.f0, wrappedContext);
+ }
+ lastFlush = System.currentTimeMillis();
+ reduceBuffer.clear();
+ }
+
+ /**
+ * Wrapper of {@link SinkWriter.Context}.
+ *
+ * <p>When records arrives, the {@link ReducingUpsertWriter} updates the current {@link
+ * SinkWriter.Context} and memorize the timestamp with the records. When flushing, the {@link
+ * ReducingUpsertWriter} will emit the records in the buffer with memorized timestamp.
+ */
+ private static class WrappedContext implements SinkWriter.Context {
+ private long timestamp;
+ private SinkWriter.Context context;
+
+ @Override
+ public long currentWatermark() {
+ checkNotNull(context, "context must be set before retrieving it.");
+ return context.currentWatermark();
+ }
+
+ @Override
+ public Long timestamp() {
+ checkNotNull(timestamp, "timestamp must to be set before retrieving it.");
+ return timestamp;
+ }
+
+ public void setTimestamp(long timestamp) {
+ this.timestamp = timestamp;
+ }
+
+ public void setContext(SinkWriter.Context context) {
+ this.context = context;
+ }
+ }
+}
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/ReducingUpsertWriterTest.java
similarity index 72%
rename from flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/BufferedUpsertSinkFunctionTest.java
rename to flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriterTest.java
index 81f6d5f..a318189 100644
--- 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/ReducingUpsertWriterTest.java
@@ -1,13 +1,12 @@
/*
- * 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
+ * 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
+ * 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,
@@ -18,16 +17,10 @@
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.ExecutionConfig;
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.api.connector.sink.Sink;
+import org.apache.flink.api.connector.sink.SinkWriter;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
@@ -43,6 +36,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import java.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
@@ -51,6 +45,7 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.function.Function;
import static org.apache.flink.types.RowKind.DELETE;
import static org.apache.flink.types.RowKind.INSERT;
@@ -58,10 +53,9 @@ 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}. */
+/** Tests for {@link ReducingUpsertWriter}. */
@RunWith(Parameterized.class)
-public class BufferedUpsertSinkFunctionTest {
-
+public class ReducingUpsertWriterTest {
@Parameterized.Parameters(name = "object reuse = {0}")
public static Object[] enableObjectReuse() {
return new Boolean[] {true, false};
@@ -137,26 +131,34 @@ public class BufferedUpsertSinkFunctionTest {
StringData.fromString("Kevin Jones"),
null,
1010,
+ TimestampData.fromInstant(Instant.parse("2021-03-30T21:00:00Z"))),
+ GenericRowData.ofKind(
+ DELETE,
+ 1005,
+ StringData.fromString("A Teaspoon of Java 1.8"),
+ StringData.fromString("Kevin Jones"),
+ null,
+ 1010,
TimestampData.fromInstant(Instant.parse("2021-03-30T21:00:00Z")))
};
private final boolean enableObjectReuse;
- public BufferedUpsertSinkFunctionTest(boolean enableObjectReuse) {
+ public ReducingUpsertWriterTest(boolean enableObjectReuse) {
this.enableObjectReuse = enableObjectReuse;
}
@Test
public void testWriteData() throws Exception {
- MockedSinkFunction sinkFunction = new MockedSinkFunction();
- BufferedUpsertSinkFunction bufferedSink = createBufferedSink(sinkFunction);
+ final MockedSinkWriter writer = new MockedSinkWriter();
+ final ReducingUpsertWriter<?> bufferedWriter = createBufferedWriter(writer);
- // write 3 records which doesn't trigger batch size
- writeData(bufferedSink, new ReusableIterator(0, 3));
- assertTrue(sinkFunction.rowDataCollectors.isEmpty());
+ // write 4 records which doesn't trigger batch size
+ writeData(bufferedWriter, new ReusableIterator(0, 4));
+ assertTrue(writer.rowDataCollectors.isEmpty());
// write one more record, and should flush the buffer
- writeData(bufferedSink, new ReusableIterator(3, 1));
+ writeData(bufferedWriter, new ReusableIterator(7, 1));
HashMap<Integer, List<RowData>> expected = new HashMap<>();
expected.put(
@@ -193,22 +195,34 @@ public class BufferedUpsertSinkFunctionTest {
55,
TimestampData.fromInstant(Instant.parse("2021-03-30T18:00:00Z")))));
- compareCompactedResult(expected, sinkFunction.rowDataCollectors);
+ expected.put(
+ 1005,
+ Collections.singletonList(
+ GenericRowData.ofKind(
+ DELETE,
+ 1005,
+ 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, writer.rowDataCollectors);
- sinkFunction.rowDataCollectors.clear();
+ writer.rowDataCollectors.clear();
// write remaining data, and they are still buffered
- writeData(bufferedSink, new ReusableIterator(4, 3));
- assertTrue(sinkFunction.rowDataCollectors.isEmpty());
+ writeData(bufferedWriter, new ReusableIterator(4, 3));
+ assertTrue(writer.rowDataCollectors.isEmpty());
}
@Test
public void testFlushDataWhenCheckpointing() throws Exception {
- MockedSinkFunction sinkFunction = new MockedSinkFunction();
- BufferedUpsertSinkFunction bufferedFunction = createBufferedSink(sinkFunction);
+ final MockedSinkWriter writer = new MockedSinkWriter();
+ final ReducingUpsertWriter<?> bufferedWriter = createBufferedWriter(writer);
// write all data, there should be 3 records are still buffered
- writeData(bufferedFunction, new ReusableIterator(0, TEST_DATA.length));
+ writeData(bufferedWriter, new ReusableIterator(0, 4));
// snapshot should flush the buffer
- bufferedFunction.snapshotState(null);
+ bufferedWriter.prepareCommit(true);
HashMap<Integer, List<RowData>> expected = new HashMap<>();
expected.put(
@@ -243,47 +257,9 @@ public class BufferedUpsertSinkFunctionTest {
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());
+ TimestampData.fromInstant(Instant.parse("2021-03-30T18:00:00Z")))));
- 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, Iterator<RowData> iterator)
- throws Exception {
- while (iterator.hasNext()) {
- RowData next = iterator.next();
- long rowtime = next.getTimestamp(TIMESTAMP_INDICES, 3).getMillisecond();
- sink.invoke(next, SinkContextUtil.forTimestamp(rowtime));
- }
+ compareCompactedResult(expected, writer.rowDataCollectors);
}
private void compareCompactedResult(
@@ -301,54 +277,82 @@ public class BufferedUpsertSinkFunctionTest {
}
}
- // --------------------------------------------------------------------------------------------
-
- private class MockedSinkFunction extends RichSinkFunction<RowData>
- implements CheckpointedFunction, CheckpointListener {
+ private void writeData(ReducingUpsertWriter<?> writer, Iterator<RowData> iterator)
+ throws Exception {
+ while (iterator.hasNext()) {
+ RowData next = iterator.next();
+ long rowtime = next.getTimestamp(TIMESTAMP_INDICES, 3).getMillisecond();
+ writer.write(
+ next,
+ new SinkWriter.Context() {
+ @Override
+ public long currentWatermark() {
+ throw new UnsupportedOperationException("Not implemented.");
+ }
+
+ @Override
+ public Long timestamp() {
+ return rowtime;
+ }
+ });
+ }
+ }
- private static final long serialVersionUID = 1L;
- private final RuntimeContext context = new MockStreamingRuntimeContext(true, 1, 1);
- transient List<RowData> rowDataCollectors;
+ @SuppressWarnings("unchecked")
+ private ReducingUpsertWriter<?> createBufferedWriter(MockedSinkWriter sinkWriter) {
+ TypeInformation<RowData> typeInformation =
+ (TypeInformation<RowData>)
+ new SinkRuntimeProviderContext(false)
+ .createTypeInformation(SCHEMA.toPhysicalRowDataType());
+ return new ReducingUpsertWriter<>(
+ sinkWriter,
+ SCHEMA.toPhysicalRowDataType(),
+ new int[] {keyIndices},
+ BUFFER_FLUSH_MODE,
+ new Sink.ProcessingTimeService() {
+ @Override
+ public long getCurrentProcessingTime() {
+ return 0;
+ }
+
+ @Override
+ public void registerProcessingTimer(
+ long time, ProcessingTimeCallback processingTimerCallback) {}
+ },
+ enableObjectReuse
+ ? typeInformation.createSerializer(new ExecutionConfig())::copy
+ : Function.identity());
+ }
- MockedSinkFunction() {
- if (enableObjectReuse) {
- context.getExecutionConfig().enableObjectReuse();
- }
- }
+ private static class MockedSinkWriter implements SinkWriter<RowData, Void, Void> {
- @Override
- public RuntimeContext getRuntimeContext() {
- return context;
- }
+ transient List<RowData> rowDataCollectors;
- @Override
- public void open(Configuration parameters) throws Exception {
- super.open(parameters);
+ MockedSinkWriter() {
rowDataCollectors = new ArrayList<>();
}
@Override
- public void notifyCheckpointComplete(long checkpointId) {
- // do nothing
+ public void write(RowData element, Context context)
+ throws IOException, InterruptedException {
+ assertEquals(
+ element.getTimestamp(TIMESTAMP_INDICES, 3).toInstant(),
+ Instant.ofEpochMilli(context.timestamp()));
+ rowDataCollectors.add(element);
}
@Override
- public void snapshotState(FunctionSnapshotContext context) {
- // do nothing
+ public List<Void> prepareCommit(boolean flush) throws IOException, InterruptedException {
+ return null;
}
@Override
- public void initializeState(FunctionInitializationContext context) {
- // do nothing
+ public List<Void> snapshotState() throws IOException {
+ return null;
}
@Override
- public void invoke(RowData value, Context context) {
- assertEquals(
- value.getTimestamp(TIMESTAMP_INDICES, 3).toInstant(),
- Instant.ofEpochMilli(context.timestamp()));
- rowDataCollectors.add(value);
- }
+ public void close() throws Exception {}
}
private class ReusableIterator implements Iterator<RowData> {
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 0eaf650..9638fd2 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
@@ -26,11 +26,13 @@ import org.apache.flink.formats.avro.AvroRowDataSerializationSchema;
import org.apache.flink.formats.avro.RowDataToAvroConverters;
import org.apache.flink.formats.avro.registry.confluent.ConfluentRegistryAvroSerializationSchema;
import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
import org.apache.flink.streaming.connectors.kafka.sink.KafkaSink;
+import org.apache.flink.streaming.runtime.operators.sink.SinkOperatorFactory;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.Column;
@@ -39,13 +41,14 @@ import org.apache.flink.table.catalog.UniqueConstraint;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
import org.apache.flink.table.connector.sink.DynamicTableSink;
-import org.apache.flink.table.connector.sink.SinkFunctionProvider;
import org.apache.flink.table.connector.sink.SinkProvider;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import org.apache.flink.table.connector.source.SourceFunctionProvider;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.factories.TestFormatFactory;
import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext;
@@ -197,6 +200,7 @@ public class UpsertKafkaDynamicTableFactoryTest extends TestLogger {
assertThat(sink, instanceOf(KafkaSink.class));
}
+ @SuppressWarnings("rawtypes")
@Test
public void testBufferedTableSink() {
// Construct table sink using options and table sink factory.
@@ -231,10 +235,23 @@ public class UpsertKafkaDynamicTableFactoryTest extends TestLogger {
// 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));
+ assertThat(provider, instanceOf(DataStreamSinkProvider.class));
+ final DataStreamSinkProvider sinkProvider = (DataStreamSinkProvider) provider;
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ sinkProvider.consumeDataStream(env.fromElements(new BinaryRowData(1)));
+ final StreamOperatorFactory<?> sinkOperatorFactory =
+ env.getStreamGraph().getStreamNodes().stream()
+ .filter(n -> n.getOperatorName().contains("Sink"))
+ .findFirst()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ "Expected operator with name Sink in stream graph."))
+ .getOperatorFactory();
+ assertThat(sinkOperatorFactory, instanceOf(SinkOperatorFactory.class));
+ assertThat(
+ ((SinkOperatorFactory) sinkOperatorFactory).getSink(),
+ instanceOf(ReducingUpsertSink.class));
}
@Test
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperatorFactory.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperatorFactory.java
index 28c294f..ad72168 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperatorFactory.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperatorFactory.java
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.runtime.operators.sink;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.connector.sink.Committer;
import org.apache.flink.api.connector.sink.Sink;
import org.apache.flink.api.connector.sink.SinkWriter;
@@ -114,4 +115,9 @@ public final class SinkOperatorFactory<InputT, CommT, WriterStateT>
public Class<? extends StreamOperator> getStreamOperatorClass(ClassLoader classLoader) {
return SinkOperator.class;
}
+
+ @VisibleForTesting
+ public Sink<InputT, CommT, WriterStateT, ?> getSink() {
+ return sink;
+ }
}
[flink] 01/02: [FLINK-23621] Add InterruptedException to
ProcessingTimeService#onProcessingTime
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 134f388323f0c2b625504fc58e874133f467f781
Author: Fabian Paul <fa...@ververica.com>
AuthorDate: Fri Aug 13 11:44:59 2021 +0200
[FLINK-23621] Add InterruptedException to ProcessingTimeService#onProcessingTime
---
.../org/apache/flink/connector/file/sink/writer/FileWriterTest.java | 6 +++---
.../src/main/java/org/apache/flink/api/connector/sink/Sink.java | 2 +-
2 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterTest.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterTest.java
index 0651031..2115f37 100644
--- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterTest.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterTest.java
@@ -192,7 +192,7 @@ public class FileWriterTest {
}
@Test
- public void testOnProcessingTime() throws IOException {
+ public void testOnProcessingTime() throws IOException, InterruptedException {
File outDir = TEMP_FOLDER.newFolder();
Path path = new Path(outDir.toURI());
@@ -327,7 +327,7 @@ public class FileWriterTest {
if (time <= now) {
try {
processingTimeCallback.onProcessingTime(now);
- } catch (IOException e) {
+ } catch (IOException | InterruptedException e) {
ExceptionUtils.rethrow(e);
}
} else {
@@ -335,7 +335,7 @@ public class FileWriterTest {
}
}
- public void advanceTo(long time) throws IOException {
+ public void advanceTo(long time) throws IOException, InterruptedException {
if (time > now) {
now = time;
diff --git a/flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java b/flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
index 99d47e9..91c4383 100644
--- a/flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
+++ b/flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
@@ -193,7 +193,7 @@ public interface Sink<InputT, CommT, WriterStateT, GlobalCommT> extends Serializ
*
* @param time The time this callback was registered for.
*/
- void onProcessingTime(long time) throws IOException;
+ void onProcessingTime(long time) throws IOException, InterruptedException;
}
}
}