You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2019/06/04 12:34:36 UTC
[flink] branch master updated (a36ec7f -> 1de854a)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from a36ec7f [hotfix][network] Drop redundant getSizeUnsafe from Buffer interface
new 70fa80e [FLINK-11249][kafka] Use custom serializers for NextTransactionalIdHint in 0.11 and universal FlinkKafkaProducer
new 9be40f6 [FLINK-11249][kafka,test] Update Kafka migration resources with new NextTransactionalIdHint serializer
new 1de854a [FLINK-11249][kafka] Fix migration from FlinkKafkaProducer0.11 to universal
The 3 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:
docs/dev/connectors/kafka.md | 9 ++
.../connectors/kafka/FlinkKafkaProducer011.java | 132 +++++++++++++++++++-
.../NextTransactionalIdHintSerializerTest.java | 56 +++++++++
...fka-migration-kafka-producer-flink-1.8-snapshot | Bin 1746 -> 2032 bytes
.../connectors/kafka/FlinkKafkaProducer.java | 136 ++++++++++++++++++++-
.../connectors/kafka/FlinkKafkaProducer011.java | 69 +++++++++++
.../FlinkKafkaProducerMigrationOperatorTest.java | 58 +++++++++
.../NextTransactionalIdHintSerializerTest.java | 56 +++++++++
....11-migration-kafka-producer-flink-1.8-snapshot | Bin 0 -> 2032 bytes
...fka-migration-kafka-producer-flink-1.8-snapshot | Bin 1731 -> 1232 bytes
10 files changed, 512 insertions(+), 4 deletions(-)
create mode 100644 flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/NextTransactionalIdHintSerializerTest.java
create mode 100644 flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
create mode 100644 flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerMigrationOperatorTest.java
create mode 100644 flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/NextTransactionalIdHintSerializerTest.java
create mode 100644 flink-connectors/flink-connector-kafka/src/test/resources/kafka-0.11-migration-kafka-producer-flink-1.8-snapshot
[flink] 03/03: [FLINK-11249][kafka] Fix migration from
FlinkKafkaProducer0.11 to universal
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1de854a7fa907c47ecbcd9aee625467e0127f929
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Feb 11 16:22:45 2019 +0100
[FLINK-11249][kafka] Fix migration from FlinkKafkaProducer0.11 to universal
Add backward compatibile classes to the universal FlinkKafkaProducer, so that it can restore from 0.11 checkpoints.
---
docs/dev/connectors/kafka.md | 9 +++
.../connectors/kafka/FlinkKafkaProducer011.java | 69 +++++++++++++++++++++
.../FlinkKafkaProducerMigrationOperatorTest.java | 58 +++++++++++++++++
....11-migration-kafka-producer-flink-1.8-snapshot | Bin 0 -> 2032 bytes
4 files changed, 136 insertions(+)
diff --git a/docs/dev/connectors/kafka.md b/docs/dev/connectors/kafka.md
index 8f893c4..4d0759f 100644
--- a/docs/dev/connectors/kafka.md
+++ b/docs/dev/connectors/kafka.md
@@ -135,6 +135,15 @@ The universal Kafka connector is compatible with older and newer Kafka brokers t
It is compatible with broker versions 0.11.0 or newer, depending on the features used.
For details on Kafka compatibility, please refer to the [Kafka documentation](https://kafka.apache.org/protocol.html#protocol_compatibility).
+### Migrating Kafka Connector from 0.11 to universal
+
+In order to perform the migration, see the [upgrading jobs and Flink versions guide]({{ site.baseurl }}/ops/upgrading.html)
+and:
+* Use Flink 1.9 or newer for the whole process.
+* Do not upgrade the Flink and operators at the same time.
+* Make sure that Kafka Consumer and/or Kafka Producer used in your job have assigned unique identifiers (`uid`):
+* Use stop with savepoint feature to take the savepoint (for example by using `stop --withSavepoint`)[CLI command]({{ site.baseurl }}/ops/cli.html).
+
### Usage
To use the universal Kafka connector add a dependency to it:
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
new file mode 100644
index 0000000..3d10a8e
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
@@ -0,0 +1,69 @@
+/*
+ * 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;
+
+import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
+
+/**
+ * Compatibility class to make migration possible from the 0.11 connector to the universal one.
+ *
+ * <p>Problem is that FlinkKafkaProducer (universal) and FlinkKafkaProducer011 have different names and
+ * they both defined static classes NextTransactionalIdHint, KafkaTransactionState and
+ * KafkaTransactionContext inside the parent classes. This is causing incompatibility problems since
+ * for example FlinkKafkaProducer011.KafkaTransactionState and FlinkKafkaProducer.KafkaTransactionState
+ * are treated as completely incompatible classes, despite being identical.
+ *
+ * <p>This issue is solved by using custom serialization logic: keeping a fake/dummy
+ * FlinkKafkaProducer011.*Serializer classes in the universal connector
+ * (this class), as entry points for the deserialization and converting them to
+ * FlinkKafkaProducer.*Serializer counter parts. After all serialized binary data are exactly
+ * the same in all of those cases.
+ *
+ * <p>For more details check FLINK-11249 and the discussion in the pull requests.
+ */
+//CHECKSTYLE:OFF: JavadocType
+public class FlinkKafkaProducer011 {
+ public static class NextTransactionalIdHintSerializer {
+ public static final class NextTransactionalIdHintSerializerSnapshot extends SimpleTypeSerializerSnapshot<FlinkKafkaProducer.NextTransactionalIdHint> {
+ public NextTransactionalIdHintSerializerSnapshot() {
+ super(FlinkKafkaProducer.NextTransactionalIdHintSerializer::new);
+ }
+ }
+ }
+
+ public static class ContextStateSerializer {
+ public static final class ContextStateSerializerSnapshot extends SimpleTypeSerializerSnapshot<FlinkKafkaProducer.KafkaTransactionContext> {
+ public ContextStateSerializerSnapshot() {
+ super(FlinkKafkaProducer.ContextStateSerializer::new);
+ }
+ }
+ }
+
+ public static class TransactionStateSerializer {
+ public static final class TransactionStateSerializerSnapshot extends SimpleTypeSerializerSnapshot<FlinkKafkaProducer.KafkaTransactionState> {
+ public TransactionStateSerializerSnapshot() {
+ super(FlinkKafkaProducer.TransactionStateSerializer::new);
+ }
+ }
+ }
+
+ public static class NextTransactionalIdHint extends FlinkKafkaProducer.NextTransactionalIdHint {
+ }
+}
+//CHECKSTYLE:ON: JavadocType
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerMigrationOperatorTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerMigrationOperatorTest.java
new file mode 100644
index 0000000..48a754d
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerMigrationOperatorTest.java
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+import org.apache.flink.testutils.migration.MigrationVersion;
+
+import org.junit.Ignore;
+import org.junit.runners.Parameterized;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+/**
+ * Migration test from FlinkKafkaProducer011 operator.
+ */
+public class FlinkKafkaProducerMigrationOperatorTest extends FlinkKafkaProducerMigrationTest {
+ @Parameterized.Parameters(name = "Migration Savepoint: {0}")
+ public static Collection<MigrationVersion> parameters() {
+ return Arrays.asList(
+ MigrationVersion.v1_8);
+ }
+
+ public FlinkKafkaProducerMigrationOperatorTest(MigrationVersion testMigrateVersion) {
+ super(testMigrateVersion);
+ }
+
+ @Override
+ public String getOperatorSnapshotPath(MigrationVersion version) {
+ return "src/test/resources/kafka-0.11-migration-kafka-producer-flink-" + version + "-snapshot";
+ }
+
+ /**
+ * This test depends on the resources generated by {@link FlinkKafkaProducer011MigrationTest}.
+ * Run {@link FlinkKafkaProducer011MigrationTest#writeSnapshot()} and copy the created resource
+ * file to the path specified by the {@link #getOperatorSnapshotPath(MigrationVersion)} method.
+ */
+ @Ignore
+ @Override
+ public void writeSnapshot() throws Exception {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/resources/kafka-0.11-migration-kafka-producer-flink-1.8-snapshot b/flink-connectors/flink-connector-kafka/src/test/resources/kafka-0.11-migration-kafka-producer-flink-1.8-snapshot
new file mode 100644
index 0000000..0a22c69
Binary files /dev/null and b/flink-connectors/flink-connector-kafka/src/test/resources/kafka-0.11-migration-kafka-producer-flink-1.8-snapshot differ
[flink] 02/03: [FLINK-11249][kafka,
test] Update Kafka migration resources with new
NextTransactionalIdHint serializer
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9be40f681d40be448dfdb925684e3d6b0d0635af
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Tue Mar 26 15:30:14 2019 +0100
[FLINK-11249][kafka,test] Update Kafka migration resources with new NextTransactionalIdHint serializer
---
.../kafka-migration-kafka-producer-flink-1.8-snapshot | Bin 1746 -> 2032 bytes
.../kafka-migration-kafka-producer-flink-1.8-snapshot | Bin 1731 -> 1232 bytes
2 files changed, 0 insertions(+), 0 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/resources/kafka-migration-kafka-producer-flink-1.8-snapshot b/flink-connectors/flink-connector-kafka-0.11/src/test/resources/kafka-migration-kafka-producer-flink-1.8-snapshot
index 29c6ccc..0a22c69 100644
Binary files a/flink-connectors/flink-connector-kafka-0.11/src/test/resources/kafka-migration-kafka-producer-flink-1.8-snapshot and b/flink-connectors/flink-connector-kafka-0.11/src/test/resources/kafka-migration-kafka-producer-flink-1.8-snapshot differ
diff --git a/flink-connectors/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.8-snapshot b/flink-connectors/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.8-snapshot
index 58c832a..b3f7632 100644
Binary files a/flink-connectors/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.8-snapshot and b/flink-connectors/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.8-snapshot differ
[flink] 01/03: [FLINK-11249][kafka] Use custom serializers for
NextTransactionalIdHint in 0.11 and universal FlinkKafkaProducer
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 70fa80e3862b367be22b593db685f9898a2838ef
Author: yanghua <ya...@gmail.com>
AuthorDate: Thu Jan 3 23:14:09 2019 +0800
[FLINK-11249][kafka] Use custom serializers for NextTransactionalIdHint in 0.11 and universal FlinkKafkaProducer
---
.../connectors/kafka/FlinkKafkaProducer011.java | 132 +++++++++++++++++++-
.../NextTransactionalIdHintSerializerTest.java | 56 +++++++++
.../connectors/kafka/FlinkKafkaProducer.java | 136 ++++++++++++++++++++-
.../NextTransactionalIdHintSerializerTest.java | 56 +++++++++
4 files changed, 376 insertions(+), 4 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
index 8b3cccd..277f891 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
@@ -174,10 +174,16 @@ public class FlinkKafkaProducer011<IN>
/**
* Descriptor of the transactional IDs list.
+ * Note: This state is serialized by Kryo Serializer and it has compatibility problem that will be removed later.
+ * Please use NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR_V2.
*/
+ @Deprecated
private static final ListStateDescriptor<NextTransactionalIdHint> NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR =
new ListStateDescriptor<>("next-transactional-id-hint", TypeInformation.of(NextTransactionalIdHint.class));
+ private static final ListStateDescriptor<NextTransactionalIdHint> NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR_V2 =
+ new ListStateDescriptor<>("next-transactional-id-hint-v2", new NextTransactionalIdHintSerializer());
+
/**
* State for nextTransactionalIdHint.
*/
@@ -816,8 +822,8 @@ public class FlinkKafkaProducer011<IN>
semantic = Semantic.NONE;
}
- nextTransactionalIdHintState = context.getOperatorStateStore().getUnionListState(
- NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR);
+ migrateNextTransactionalIdHindState(context);
+
transactionalIdsGenerator = new TransactionalIdsGenerator(
getRuntimeContext().getTaskName() + "-" + ((StreamingRuntimeContext) getRuntimeContext()).getOperatorUniqueID(),
getRuntimeContext().getIndexOfThisSubtask(),
@@ -1002,6 +1008,19 @@ public class FlinkKafkaProducer011<IN>
in.defaultReadObject();
}
+ private void migrateNextTransactionalIdHindState(FunctionInitializationContext context) throws Exception {
+ ListState<NextTransactionalIdHint> oldNextTransactionalIdHintState = context.getOperatorStateStore().getUnionListState(
+ NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR);
+ nextTransactionalIdHintState = context.getOperatorStateStore().getUnionListState(NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR_V2);
+
+ ArrayList<NextTransactionalIdHint> oldTransactionalIdHints = Lists.newArrayList(oldNextTransactionalIdHintState.get());
+ if (!oldTransactionalIdHints.isEmpty()) {
+ nextTransactionalIdHintState.addAll(oldTransactionalIdHints);
+ //clear old state
+ oldNextTransactionalIdHintState.clear();
+ }
+ }
+
private static Properties getPropertiesFromBrokerList(String brokerList) {
String[] elements = brokerList.split(",");
@@ -1355,5 +1374,114 @@ public class FlinkKafkaProducer011<IN>
this.lastParallelism = parallelism;
this.nextFreeTransactionalId = nextFreeTransactionalId;
}
+
+ @Override
+ public String toString() {
+ return "NextTransactionalIdHint[" +
+ "lastParallelism=" + lastParallelism +
+ ", nextFreeTransactionalId=" + nextFreeTransactionalId +
+ ']';
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ NextTransactionalIdHint that = (NextTransactionalIdHint) o;
+
+ if (lastParallelism != that.lastParallelism) {
+ return false;
+ }
+ return nextFreeTransactionalId == that.nextFreeTransactionalId;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = lastParallelism;
+ result = 31 * result + (int) (nextFreeTransactionalId ^ (nextFreeTransactionalId >>> 32));
+ return result;
+ }
+ }
+
+ /**
+ * {@link org.apache.flink.api.common.typeutils.TypeSerializer} for
+ * {@link NextTransactionalIdHint}.
+ */
+ @VisibleForTesting
+ @Internal
+ public static class NextTransactionalIdHintSerializer extends TypeSerializerSingleton<NextTransactionalIdHint> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean isImmutableType() {
+ return true;
+ }
+
+ @Override
+ public NextTransactionalIdHint createInstance() {
+ return new NextTransactionalIdHint();
+ }
+
+ @Override
+ public NextTransactionalIdHint copy(NextTransactionalIdHint from) {
+ return from;
+ }
+
+ @Override
+ public NextTransactionalIdHint copy(NextTransactionalIdHint from, NextTransactionalIdHint reuse) {
+ return from;
+ }
+
+ @Override
+ public int getLength() {
+ return Long.BYTES + Integer.BYTES;
+ }
+
+ @Override
+ public void serialize(NextTransactionalIdHint record, DataOutputView target) throws IOException {
+ target.writeLong(record.nextFreeTransactionalId);
+ target.writeInt(record.lastParallelism);
+ }
+
+ @Override
+ public NextTransactionalIdHint deserialize(DataInputView source) throws IOException {
+ long nextFreeTransactionalId = source.readLong();
+ int lastParallelism = source.readInt();
+ return new NextTransactionalIdHint(lastParallelism, nextFreeTransactionalId);
+ }
+
+ @Override
+ public NextTransactionalIdHint deserialize(NextTransactionalIdHint reuse, DataInputView source) throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ target.writeLong(source.readLong());
+ target.writeInt(source.readInt());
+ }
+
+ @Override
+ public TypeSerializerSnapshot<NextTransactionalIdHint> snapshotConfiguration() {
+ return new NextTransactionalIdHintSerializerSnapshot();
+ }
+
+ /**
+ * Serializer configuration snapshot for compatibility and format evolution.
+ */
+ @SuppressWarnings("WeakerAccess")
+ public static final class NextTransactionalIdHintSerializerSnapshot extends SimpleTypeSerializerSnapshot<NextTransactionalIdHint> {
+
+ public NextTransactionalIdHintSerializerSnapshot() {
+ super(NextTransactionalIdHintSerializer::new);
+ }
+ }
}
+
}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/NextTransactionalIdHintSerializerTest.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/NextTransactionalIdHintSerializerTest.java
new file mode 100644
index 0000000..db9bbc6
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/NextTransactionalIdHintSerializerTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+
+/**
+ * A test for the {@link TypeSerializer TypeSerializers} used for
+ * {@link FlinkKafkaProducer011.NextTransactionalIdHint}.
+ */
+public class NextTransactionalIdHintSerializerTest extends
+ SerializerTestBase<FlinkKafkaProducer011.NextTransactionalIdHint> {
+
+ @Override
+ protected TypeSerializer<FlinkKafkaProducer011.NextTransactionalIdHint> createSerializer() {
+ return new FlinkKafkaProducer011.NextTransactionalIdHintSerializer();
+ }
+
+ @Override
+ protected int getLength() {
+ return Long.BYTES + Integer.BYTES;
+ }
+
+ @Override
+ protected Class<FlinkKafkaProducer011.NextTransactionalIdHint> getTypeClass() {
+ return (Class) FlinkKafkaProducer011.NextTransactionalIdHint.class;
+ }
+
+ @Override
+ protected FlinkKafkaProducer011.NextTransactionalIdHint[] getTestData() {
+ return new FlinkKafkaProducer011.NextTransactionalIdHint[] {
+ new FlinkKafkaProducer011.NextTransactionalIdHint(1, 0L),
+ new FlinkKafkaProducer011.NextTransactionalIdHint(1, 1L),
+ new FlinkKafkaProducer011.NextTransactionalIdHint(1, -1L),
+ new FlinkKafkaProducer011.NextTransactionalIdHint(2, 0L),
+ new FlinkKafkaProducer011.NextTransactionalIdHint(2, 1L),
+ new FlinkKafkaProducer011.NextTransactionalIdHint(2, -1L),
+ };
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
index 9eb2df8..3ab6a06 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
@@ -176,9 +176,15 @@ public class FlinkKafkaProducer<IN>
/**
* Descriptor of the transactional IDs list.
+ * Note: This state is serialized by Kryo Serializer and it has compatibility problem that will be removed later.
+ * Please use NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR_V2.
*/
+ @Deprecated
private static final ListStateDescriptor<FlinkKafkaProducer.NextTransactionalIdHint> NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR =
- new ListStateDescriptor<>("next-transactional-id-hint", TypeInformation.of(FlinkKafkaProducer.NextTransactionalIdHint.class));
+ new ListStateDescriptor<>("next-transactional-id-hint", TypeInformation.of(NextTransactionalIdHint.class));
+
+ private static final ListStateDescriptor<FlinkKafkaProducer.NextTransactionalIdHint> NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR_V2 =
+ new ListStateDescriptor<>("next-transactional-id-hint-v2", new NextTransactionalIdHintSerializer());
/**
* State for nextTransactionalIdHint.
@@ -819,7 +825,12 @@ public class FlinkKafkaProducer<IN>
}
nextTransactionalIdHintState = context.getOperatorStateStore().getUnionListState(
- NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR);
+ NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR_V2);
+
+ if (context.getOperatorStateStore().getRegisteredStateNames().contains(NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR)) {
+ migrateNextTransactionalIdHindState(context);
+ }
+
transactionalIdsGenerator = new TransactionalIdsGenerator(
getRuntimeContext().getTaskName() + "-" + ((StreamingRuntimeContext) getRuntimeContext()).getOperatorUniqueID(),
getRuntimeContext().getIndexOfThisSubtask(),
@@ -1008,6 +1019,19 @@ public class FlinkKafkaProducer<IN>
in.defaultReadObject();
}
+ private void migrateNextTransactionalIdHindState(FunctionInitializationContext context) throws Exception {
+ ListState<NextTransactionalIdHint> oldNextTransactionalIdHintState = context.getOperatorStateStore().getUnionListState(
+ NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR);
+ nextTransactionalIdHintState = context.getOperatorStateStore().getUnionListState(NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR_V2);
+
+ ArrayList<NextTransactionalIdHint> oldTransactionalIdHints = Lists.newArrayList(oldNextTransactionalIdHintState.get());
+ if (!oldTransactionalIdHints.isEmpty()) {
+ nextTransactionalIdHintState.addAll(oldTransactionalIdHints);
+ //clear old state
+ oldNextTransactionalIdHintState.clear();
+ }
+ }
+
private static Properties getPropertiesFromBrokerList(String brokerList) {
String[] elements = brokerList.split(",");
@@ -1363,5 +1387,113 @@ public class FlinkKafkaProducer<IN>
this.lastParallelism = parallelism;
this.nextFreeTransactionalId = nextFreeTransactionalId;
}
+
+ @Override
+ public String toString() {
+ return "NextTransactionalIdHint[" +
+ "lastParallelism=" + lastParallelism +
+ ", nextFreeTransactionalId=" + nextFreeTransactionalId +
+ ']';
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ NextTransactionalIdHint that = (NextTransactionalIdHint) o;
+
+ if (lastParallelism != that.lastParallelism) {
+ return false;
+ }
+ return nextFreeTransactionalId == that.nextFreeTransactionalId;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = lastParallelism;
+ result = 31 * result + (int) (nextFreeTransactionalId ^ (nextFreeTransactionalId >>> 32));
+ return result;
+ }
+ }
+
+ /**
+ * {@link org.apache.flink.api.common.typeutils.TypeSerializer} for
+ * {@link FlinkKafkaProducer.NextTransactionalIdHint}.
+ */
+ @VisibleForTesting
+ @Internal
+ public static class NextTransactionalIdHintSerializer extends TypeSerializerSingleton<NextTransactionalIdHint> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean isImmutableType() {
+ return true;
+ }
+
+ @Override
+ public NextTransactionalIdHint createInstance() {
+ return new NextTransactionalIdHint();
+ }
+
+ @Override
+ public NextTransactionalIdHint copy(NextTransactionalIdHint from) {
+ return from;
+ }
+
+ @Override
+ public NextTransactionalIdHint copy(NextTransactionalIdHint from, NextTransactionalIdHint reuse) {
+ return from;
+ }
+
+ @Override
+ public int getLength() {
+ return Long.BYTES + Integer.BYTES;
+ }
+
+ @Override
+ public void serialize(NextTransactionalIdHint record, DataOutputView target) throws IOException {
+ target.writeLong(record.nextFreeTransactionalId);
+ target.writeInt(record.lastParallelism);
+ }
+
+ @Override
+ public NextTransactionalIdHint deserialize(DataInputView source) throws IOException {
+ long nextFreeTransactionalId = source.readLong();
+ int lastParallelism = source.readInt();
+ return new NextTransactionalIdHint(lastParallelism, nextFreeTransactionalId);
+ }
+
+ @Override
+ public NextTransactionalIdHint deserialize(NextTransactionalIdHint reuse, DataInputView source) throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ target.writeLong(source.readLong());
+ target.writeInt(source.readInt());
+ }
+
+ @Override
+ public TypeSerializerSnapshot<NextTransactionalIdHint> snapshotConfiguration() {
+ return new NextTransactionalIdHintSerializerSnapshot();
+ }
+
+ /**
+ * Serializer configuration snapshot for compatibility and format evolution.
+ */
+ @SuppressWarnings("WeakerAccess")
+ public static final class NextTransactionalIdHintSerializerSnapshot extends SimpleTypeSerializerSnapshot<NextTransactionalIdHint> {
+
+ public NextTransactionalIdHintSerializerSnapshot() {
+ super(NextTransactionalIdHintSerializer::new);
+ }
+ }
}
}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/NextTransactionalIdHintSerializerTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/NextTransactionalIdHintSerializerTest.java
new file mode 100644
index 0000000..e21afc8
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/NextTransactionalIdHintSerializerTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+
+/**
+ * A test for the {@link TypeSerializer TypeSerializers} used for
+ * {@link FlinkKafkaProducer.NextTransactionalIdHint}.
+ */
+public class NextTransactionalIdHintSerializerTest extends
+ SerializerTestBase<FlinkKafkaProducer.NextTransactionalIdHint> {
+
+ @Override
+ protected TypeSerializer<FlinkKafkaProducer.NextTransactionalIdHint> createSerializer() {
+ return new FlinkKafkaProducer.NextTransactionalIdHintSerializer();
+ }
+
+ @Override
+ protected int getLength() {
+ return Long.BYTES + Integer.BYTES;
+ }
+
+ @Override
+ protected Class<FlinkKafkaProducer.NextTransactionalIdHint> getTypeClass() {
+ return (Class) FlinkKafkaProducer.NextTransactionalIdHint.class;
+ }
+
+ @Override
+ protected FlinkKafkaProducer.NextTransactionalIdHint[] getTestData() {
+ return new FlinkKafkaProducer.NextTransactionalIdHint[] {
+ new FlinkKafkaProducer.NextTransactionalIdHint(1, 0L),
+ new FlinkKafkaProducer.NextTransactionalIdHint(1, 1L),
+ new FlinkKafkaProducer.NextTransactionalIdHint(1, -1L),
+ new FlinkKafkaProducer.NextTransactionalIdHint(2, 0L),
+ new FlinkKafkaProducer.NextTransactionalIdHint(2, 1L),
+ new FlinkKafkaProducer.NextTransactionalIdHint(2, -1L),
+ };
+ }
+}