FLIP-573: [DRAFT] Queues for Kafka

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/share/ShareAckCommittable.java

@@ -0,0 +1,123 @@
+/*
+ * 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.connector.kafka.share;
+
+import org.apache.flink.annotation.Internal;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+@Internal
+public class ShareAckCommittable implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private final long checkpointId;
+    private final String transactionalId;
+    private final long transactionOwnerId;
+    private final short transactionOwnerEpoch;
+    private final String groupId;
+    private final int sourceSubtaskId;
+
+    public ShareAckCommittable(
+            long checkpointId,
+            String transactionalId,
+            long transactionOwnerId,
+            short transactionOwnerEpoch,
+            String groupId,
+            int sourceSubtaskId) {
+        this.checkpointId = checkpointId;
+        this.transactionalId = transactionalId;
+        this.transactionOwnerId = transactionOwnerId;
+        this.transactionOwnerEpoch = transactionOwnerEpoch;
+        this.groupId = groupId;
+        this.sourceSubtaskId = sourceSubtaskId;
+    }
+
+    public long getCheckpointId() {
+        return checkpointId;
+    }
+
+    public String getTransactionalId() {
+        return transactionalId;
+    }
+
+    public long getTransactionOwnerId() {
+        return transactionOwnerId;
+    }
+
+    public short getTransactionOwnerEpoch() {
+        return transactionOwnerEpoch;
+    }
+
+    public String getGroupId() {
+        return groupId;
+    }
+
+    public int getSourceSubtaskId() {
+        return sourceSubtaskId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ShareAckCommittable that = (ShareAckCommittable) o;
+        return checkpointId == that.checkpointId
+                && transactionOwnerId == that.transactionOwnerId
+                && transactionOwnerEpoch == that.transactionOwnerEpoch
+                && sourceSubtaskId == that.sourceSubtaskId
+                && transactionalId.equals(that.transactionalId)
+                && groupId.equals(that.groupId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(
+                checkpointId,
+                transactionalId,
+                transactionOwnerId,
+                transactionOwnerEpoch,
+                groupId,
+                sourceSubtaskId);
+    }
+
+    @Override
+    public String toString() {
+        return "ShareAckCommittable{"
+                + "checkpointId="
+                + checkpointId
+                + ", transactionalId='"
+                + transactionalId
+                + '\''
+                + ", transactionOwnerId="
+                + transactionOwnerId
+                + ", transactionOwnerEpoch="
+                + transactionOwnerEpoch
+                + ", groupId='"
+                + groupId
+                + '\''
+                + ", sourceSubtaskId="
+                + sourceSubtaskId
+                + '}';
+    }
+}

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaShareEosCommittable.java

@@ -0,0 +1,125 @@
+/*
+ * 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.connector.kafka.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.kafka.share.ShareAckCommittable;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+@Internal
+public class KafkaShareEosCommittable {
+
+    public enum CommitPhase {
+        READY,
+        SINK_COMMITTED
+    }
+
+    private final long checkpointId;
+    private final List<KafkaCommittable> kafkaCommittables;
+    private final List<ShareAckCommittable> shareAckCommittables;
+    private final CommitPhase commitPhase;
+
+    public KafkaShareEosCommittable(
+            long checkpointId,
+            Collection<KafkaCommittable> kafkaCommittables,
+            Collection<ShareAckCommittable> shareAckCommittables,
+            CommitPhase commitPhase) {
+        this.checkpointId = checkpointId;
+        this.kafkaCommittables = copy(kafkaCommittables);
+        this.shareAckCommittables = copy(shareAckCommittables);
+        this.commitPhase = commitPhase;
+    }
+
+    public static KafkaShareEosCommittable ready(
+            long checkpointId,
+            Collection<KafkaCommittable> kafkaCommittables,
+            Collection<ShareAckCommittable> shareAckCommittables) {
+        return new KafkaShareEosCommittable(
+                checkpointId, kafkaCommittables, shareAckCommittables, CommitPhase.READY);
+    }
+
+    private static <T> List<T> copy(Collection<T> values) {
+        return Collections.unmodifiableList(new ArrayList<>(values));
+    }
+
+    public long getCheckpointId() {
+        return checkpointId;
+    }
+
+    public List<KafkaCommittable> getKafkaCommittables() {
+        return kafkaCommittables;
+    }
+
+    public List<ShareAckCommittable> getShareAckCommittables() {
+        return shareAckCommittables;
+    }
+
+    public CommitPhase getCommitPhase() {
+        return commitPhase;
+    }
+
+    public KafkaShareEosCommittable withSinkCommitted() {
+        if (commitPhase == CommitPhase.SINK_COMMITTED) {
+            return this;
+        }
+        return new KafkaShareEosCommittable(
+                checkpointId,
+                kafkaCommittables,
+                shareAckCommittables,
+                CommitPhase.SINK_COMMITTED);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        KafkaShareEosCommittable that = (KafkaShareEosCommittable) o;
+        return checkpointId == that.checkpointId
+                && kafkaCommittables.equals(that.kafkaCommittables)
+                && shareAckCommittables.equals(that.shareAckCommittables)
+                && commitPhase == that.commitPhase;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(checkpointId, kafkaCommittables, shareAckCommittables, commitPhase);
+    }
+
+    @Override
+    public String toString() {
+        return "KafkaShareEosCommittable{"
+                + "checkpointId="
+                + checkpointId
+                + ", kafkaCommittables="
+                + kafkaCommittables
+                + ", shareAckCommittables="
+                + shareAckCommittables
+                + ", commitPhase="
+                + commitPhase
+                + '}';
+    }
+}

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaShareEosCommittableSerializer.java

@@ -0,0 +1,106 @@
+/*
+ * 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.connector.kafka.sink;
+
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.connector.kafka.share.ShareAckCommittable;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+class KafkaShareEosCommittableSerializer
+        implements SimpleVersionedSerializer<KafkaShareEosCommittable> {
+
+    private static final KafkaCommittableSerializer KAFKA_COMMITTABLE_SERIALIZER =
+            new KafkaCommittableSerializer();
+
+    @Override
+    public int getVersion() {
+        return 1;
+    }
+
+    @Override
+    public byte[] serialize(KafkaShareEosCommittable committable) throws IOException {
+        try (final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+                final DataOutputStream out = new DataOutputStream(baos)) {
+            out.writeLong(committable.getCheckpointId());
+            out.writeInt(committable.getCommitPhase().ordinal());
+            out.writeInt(committable.getKafkaCommittables().size());
+            for (KafkaCommittable kafkaCommittable : committable.getKafkaCommittables()) {
+                byte[] bytes = KAFKA_COMMITTABLE_SERIALIZER.serialize(kafkaCommittable);
+                out.writeInt(bytes.length);
+                out.write(bytes);
+            }
+            out.writeInt(committable.getShareAckCommittables().size());
+            for (ShareAckCommittable shareAckCommittable :
+                    committable.getShareAckCommittables()) {
+                out.writeLong(shareAckCommittable.getCheckpointId());
+                out.writeUTF(shareAckCommittable.getTransactionalId());
+                out.writeLong(shareAckCommittable.getTransactionOwnerId());
+                out.writeShort(shareAckCommittable.getTransactionOwnerEpoch());
+                out.writeUTF(shareAckCommittable.getGroupId());
+                out.writeInt(shareAckCommittable.getSourceSubtaskId());
+            }
+            out.flush();
+            return baos.toByteArray();
+        }
+    }
+
+    @Override
+    public KafkaShareEosCommittable deserialize(int version, byte[] serialized)
+            throws IOException {
+        if (version > getVersion()) {
+            throw new IOException("Unknown version: " + version);
+        }
+
+        try (final ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
+                final DataInputStream in = new DataInputStream(bais)) {
+            long checkpointId = in.readLong();
+            KafkaShareEosCommittable.CommitPhase phase =
+                    KafkaShareEosCommittable.CommitPhase.values()[in.readInt()];
+            List<KafkaCommittable> kafkaCommittables = new ArrayList<>();
+            int kafkaCommittablesSize = in.readInt();
+            for (int i = 0; i < kafkaCommittablesSize; i++) {
+                byte[] bytes = new byte[in.readInt()];
+                in.readFully(bytes);
+                kafkaCommittables.add(
+                        KAFKA_COMMITTABLE_SERIALIZER.deserialize(
+                                KAFKA_COMMITTABLE_SERIALIZER.getVersion(), bytes));
+            }
+            List<ShareAckCommittable> shareAckCommittables = new ArrayList<>();
+            int shareAckCommittablesSize = in.readInt();
+            for (int i = 0; i < shareAckCommittablesSize; i++) {
+                shareAckCommittables.add(
+                        new ShareAckCommittable(
+                                in.readLong(),
+                                in.readUTF(),
+                                in.readLong(),
+                                in.readShort(),
+                                in.readUTF(),
+                                in.readInt()));
+            }
+            return new KafkaShareEosCommittable(
+                    checkpointId, kafkaCommittables, shareAckCommittables, phase);
+        }
+    }
+}