[FLINK-39813][Kinesis/Connector] Add lineage support to Kinesis Streams connector

flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/lineage/KinesisDatasetFacet.java

@@ -0,0 +1,81 @@
+/*
+ * 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.kinesis.lineage;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.streaming.api.lineage.LineageDatasetFacet;
+
+import java.util.Objects;
+
+/**
+ * Dataset facet containing Kinesis-specific metadata for lineage reporting.
+ *
+ * <p>Includes the stream ARN, region, and stream name for full identification.
+ */
+@PublicEvolving
+public class KinesisDatasetFacet implements LineageDatasetFacet {
+
+    public static final String KINESIS_FACET_NAME = "kinesis";
+
+    private final String streamArn;
+    private final String streamName;
+    private final String region;
+
+    public KinesisDatasetFacet(String streamArn, String streamName, String region) {
+        this.streamArn = streamArn;
+        this.streamName = streamName;
+        this.region = region;
+    }
+
+    public String getStreamArn() {
+        return streamArn;
+    }
+
+    public String getStreamName() {
+        return streamName;
+    }
+
+    public String getRegion() {
+        return region;
+    }
+
+    @Override
+    public String name() {
+        return KINESIS_FACET_NAME;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        KinesisDatasetFacet that = (KinesisDatasetFacet) o;
+        return Objects.equals(streamArn, that.streamArn)
+                && Objects.equals(streamName, that.streamName)
+                && Objects.equals(region, that.region);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(streamArn, streamName, region);
+    }
+}

flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/lineage/KinesisLineageUtil.java

@@ -0,0 +1,97 @@
+/*
+ * 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.kinesis.lineage;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.streaming.api.lineage.DefaultLineageDataset;
+import org.apache.flink.streaming.api.lineage.DefaultLineageVertex;
+import org.apache.flink.streaming.api.lineage.DefaultSourceLineageVertex;
+import org.apache.flink.streaming.api.lineage.LineageDataset;
+import org.apache.flink.streaming.api.lineage.LineageDatasetFacet;
+import org.apache.flink.streaming.api.lineage.LineageVertex;
+import org.apache.flink.streaming.api.lineage.SourceLineageVertex;
+
+import software.amazon.awssdk.arns.Arn;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/** Utility class for constructing lineage datasets and vertices for Kinesis Streams. */
+@Internal
+public class KinesisLineageUtil {
+
+    /** Constructs the namespace for a Kinesis stream from its ARN. */
+    public static String namespaceOf(String streamArn) {
+        // Use the ARN prefix up to the resource as namespace
+        // e.g., arn:aws:kinesis:us-west-2:123456789012:stream
+        Arn arn = Arn.fromString(streamArn);
+        return String.format(
+                "arn:%s:kinesis:%s:%s:stream",
+                arn.partition(), arn.region().orElse(""), arn.accountId().orElse(""));
+    }
+
+    /** Extracts the stream name from a Kinesis stream ARN. */
+    public static String streamNameOf(String streamArn) {
+        Arn arn = Arn.fromString(streamArn);
+        return arn.resource().resource();
+    }
+
+    public static LineageDataset datasetOf(String streamArn) {
+        return new DefaultLineageDataset(
+                streamNameOf(streamArn), namespaceOf(streamArn), Collections.emptyMap());
+    }
+
+    public static LineageDataset datasetOf(
+            String streamArn, Map<String, LineageDatasetFacet> facets) {
+        return new DefaultLineageDataset(streamNameOf(streamArn), namespaceOf(streamArn), facets);
+    }
+
+    public static LineageDataset datasetOf(
+            String streamArn, KinesisDatasetFacet kinesisDatasetFacet) {
+        Map<String, LineageDatasetFacet> facets = new HashMap<>();
+        facets.put(KinesisDatasetFacet.KINESIS_FACET_NAME, kinesisDatasetFacet);
+        return new DefaultLineageDataset(streamNameOf(streamArn), namespaceOf(streamArn), facets);
+    }
+
+    public static LineageDataset datasetOf(
+            String streamArn,
+            KinesisDatasetFacet kinesisDatasetFacet,
+            TypeDatasetFacet typeDatasetFacet) {
+        Map<String, LineageDatasetFacet> facets = new HashMap<>();
+        facets.put(KinesisDatasetFacet.KINESIS_FACET_NAME, kinesisDatasetFacet);
+        facets.put(TypeDatasetFacet.TYPE_FACET_NAME, typeDatasetFacet);
+        return new DefaultLineageDataset(streamNameOf(streamArn), namespaceOf(streamArn), facets);
+    }
+
+    public static SourceLineageVertex sourceLineageVertexOf(Collection<LineageDataset> datasets) {
+        DefaultSourceLineageVertex vertex =
+                new DefaultSourceLineageVertex(
+                        org.apache.flink.api.connector.source.Boundedness.CONTINUOUS_UNBOUNDED);
+        datasets.forEach(vertex::addDataset);
+        return vertex;
+    }
+
+    public static LineageVertex sinkLineageVertexOf(Collection<LineageDataset> datasets) {
+        DefaultLineageVertex vertex = new DefaultLineageVertex();
+        datasets.forEach(vertex::addLineageDataset);
+        return vertex;
+    }
+}

flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/lineage/TypeDatasetFacet.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.connector.kinesis.lineage;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.lineage.LineageDatasetFacet;
+
+import java.util.Objects;
+
+/**
+ * Dataset facet containing type/schema information for lineage reporting.
+ *
+ * <p>Wraps the Flink {@link TypeInformation} which contains column names and types when available
+ * (e.g., from RowType schemas).
+ */
+@PublicEvolving
+public class TypeDatasetFacet implements LineageDatasetFacet {
+
+    public static final String TYPE_FACET_NAME = "type";
+
+    private final TypeInformation<?> typeInformation;
+
+    public TypeDatasetFacet(TypeInformation<?> typeInformation) {
+        this.typeInformation = typeInformation;
+    }
+
+    public TypeInformation<?> getTypeInformation() {
+        return typeInformation;
+    }
+
+    @Override
+    public String name() {
+        return TYPE_FACET_NAME;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        TypeDatasetFacet that = (TypeDatasetFacet) o;
+        return Objects.equals(typeInformation, that.typeInformation);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(typeInformation);
+    }
+}

flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSink.java

@@ -67,7 +67,8 @@
  * @param <InputT> Type of the elements handled by this sink
  */
 @PublicEvolving
-public class KinesisStreamsSink<InputT> extends AsyncSinkBase<InputT, PutRecordsRequestEntry> {
+public class KinesisStreamsSink<InputT> extends AsyncSinkBase<InputT, PutRecordsRequestEntry>
+        implements org.apache.flink.streaming.api.lineage.LineageVertexProvider {
 
     private final boolean failOnError;
     private final String streamName;
@@ -174,4 +175,24 @@ public StatefulSinkWriter<InputT, BufferedRequestState<PutRecordsRequestEntry>>
                 kinesisClientProperties,
                 recoveredState);
     }
+
+    // ---- Lineage support ----
+
+    @Override
+    public org.apache.flink.streaming.api.lineage.LineageVertex getLineageVertex() {
+        if (streamArn != null) {
+            return org.apache.flink.connector.kinesis.lineage.KinesisLineageUtil
+                    .sinkLineageVertexOf(
+                            java.util.Collections.singletonList(
+                                    org.apache.flink.connector.kinesis.lineage.KinesisLineageUtil
+                                            .datasetOf(streamArn)));
+        }
+        // Fallback when only streamName is provided (no ARN)
+        return org.apache.flink.connector.kinesis.lineage.KinesisLineageUtil.sinkLineageVertexOf(
+                java.util.Collections.singletonList(
+                        new org.apache.flink.streaming.api.lineage.DefaultLineageDataset(
+                                streamName,
+                                "kinesis://unknown-region",
+                                java.util.Collections.emptyMap())));
+    }
 }

flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/KinesisStreamsSource.java

@@ -105,7 +105,8 @@
  */
 @Experimental
 public class KinesisStreamsSource<T>
-        implements Source<T, KinesisShardSplit, KinesisStreamsSourceEnumeratorState> {
+        implements Source<T, KinesisShardSplit, KinesisStreamsSourceEnumeratorState>,
+                org.apache.flink.streaming.api.lineage.LineageVertexProvider {
 
     private final String streamArn;
     private final Configuration sourceConfig;
@@ -354,4 +355,14 @@ private StandardRetryStrategy.Builder createExpBackoffRetryStrategyBuilder(
                 .retryOnExceptionOrCauseInstanceOf(LimitExceededException.class)
                 .maxAttempts(maxAttempts);
     }
+
+    // ---- Lineage support ----
+
+    @Override
+    public org.apache.flink.streaming.api.lineage.SourceLineageVertex getLineageVertex() {
+        return org.apache.flink.connector.kinesis.lineage.KinesisLineageUtil.sourceLineageVertexOf(
+                java.util.Collections.singletonList(
+                        org.apache.flink.connector.kinesis.lineage.KinesisLineageUtil.datasetOf(
+                                streamArn)));
+    }
 }