[FLINK-34466] Lineage interfaces for kafka connector #130

pawel-big-lebowski · 2024-10-14T11:09:38Z

FLINK-31275 is aiming to provide native lineage support in Flink's codebase with custom job listeners that get notified about job state changes as well as lineage graph extracted. As a part of that, lineage interfaces introduced in FLINK-33210 need to be implemented on the connectors' side for sources and sinks to expose lineage metadata about input and output datasets of job runs.

https://issues.apache.org/jira/browse/FLINK-34466

boring-cyborg · 2024-10-14T11:09:43Z

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

HuangZhenQiu · 2024-10-15T08:58:23Z

...ector-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/LineageFacetProvider.java

+ * Contains method which can be used for lineage schema facet extraction. Useful for classes like
+ * topic selectors or serialization schemas to extract dataset information from.
+ */
+public interface LineageFacetProvider {


I feel this interface can be moved to flink core repo.

I think so too. To me, it makes more sense to add it first into flink-core and remove here later when upgrading flink-core for flink-connector-kafka. This shouldn't block lineage interface implementation.

HuangZhenQiu · 2024-10-15T09:01:20Z

@pawel-big-lebowski
Would you please change the PR title with the jira ticket prefix. Please also enhance the PR summary with the original template.

AHeise

Overall approach looks good. I challenged a central piece of transferring the information via facets to the sink/source, so I haven't checked the tests yet. PTAL.

AHeise · 2024-10-15T08:54:00Z

...ector-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/LineageFacetProvider.java

+ * Contains method which can be used for lineage schema facet extraction. Useful for classes like
+ * topic selectors or serialization schemas to extract dataset information from.
+ */
+public interface LineageFacetProvider {


Should this be part of flink-core in the future?

#130 (comment)

AHeise · 2024-10-15T08:54:29Z

...ector-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/LineageFacetProvider.java

+     *
+     * @return
+     */
+    List<LineageDatasetFacet> getDatasetFacets();


nit: is Collection sufficient?

flink-core lineage interfaces like LineageVertex and LineageGraph also have lists, but Collection should be enough.

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/LineageUtil.java

AHeise · 2024-10-15T08:58:55Z

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/LineageUtil.java

+                    facets.stream().filter(f -> !f.equals(topicList)).collect(Collectors.toList());
+
+            topicList.get().topics.stream()
+                    .forEach(t -> datasets.add(datasetOf(namespace, t, facetsWithoutTopicList)));


nit: If you use functional style, forEach + add is rather an anti-pattern. You'd instead chain Streams and materialize them at the very end with a Collector.

AHeise · 2024-10-15T09:07:20Z

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/LineageUtil.java

+     * @param facets
+     * @return
+     */
+    public static List<LineageDataset> datasetsFrom(


This whole information flow around the facets looks a bit unclean to me.
Both Source/Sink throw a bunch of information into a list of LineageDatasetFacet, then this method is applied to take that list apart and construct the actually intended LineageDataset. So we first deliberately lose the information of what the facets are about and then we need to use a lot of (hidden) if-else to extract that information again.

WDYT of replacing the List<LineageDatasetFacet> instead with a value class that contains all relevant information:

class KafkaFacet { @Nullable String topicPattern; @Nullable List<String> topicList; Properties properties; @Nullable TypeInformation typeInformation; }

Then you can access all the different pieces of information without the isInstance/cast pattern that you use.
You can then in this method still turn all the pieces of information into separate facets.

The difficulty of this approach is that KafkaFacet properties are collected in different classes and this is currently done with LineageFacetProvider having a method Collection<LineageDatasetFacet> getDatasetFacets().

A solution to this would be to create KafkaFacetProvider interfaces (instead of LineageFacetProvider) with a method:

void buildKafkaFacet(KafkaFacetBuilder builder)

This would pass Kafka facet builder as an argument and let facet being enriched within the method calls.

@AHeise Is this something you had on your mind?

Hm I have not fully understood from which classes we actually need to extract the various data points. Could we recap here?

Source/Sink gives us the properties directly

Source gives us the type information directly but we also try to extract it from the deserialization schema (why?).

KafkaSubscriber of the source either gives us a topicPattern or a topicList.

SerializationSchema of the sink gives us the topicList

In the end, we emit a lineageVertex that has facets per topic (pattern) in some cross-product fashion. I have not fully understood how a given input looks fully expanded after datasetsFrom. Maybe you could summarize that.

Anyways, it feels like the KafkaFacet contains a list of topics that is filled through polymorphism and some parts that are filled statically. Can we maybe separate that? Would we be able to say that the topic selector/subscriber just return a list of facet names and we use them to create the facets with the statically set properties and type information?

AHeise · 2024-10-15T09:09:11Z

...afka/src/main/java/org/apache/flink/connector/kafka/lineage/facets/KafkaPropertiesFacet.java

+public class KafkaPropertiesFacet implements LineageDatasetFacet {
+
+    public static final String KAFKA_PROPERTIES_FACET_NAME = "kafkaProperties";
+    public Properties properties;


What assumptions do we make about the mutability and thread-safety of the facades? Do we nned to make defensive copies of the mutable information such as the Properties?

don't know the answers to those questions. I think it's safer to create new properties object.

AHeise · 2024-10-15T09:13:48Z

...c/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java

+        @Override
+        public List<LineageDatasetFacet> getDatasetFacets() {
+            List<LineageDatasetFacet> facets = new ArrayList<>();
+            facets.add(new KafkaTopicListFacet(Arrays.asList(topicSelector.apply(null))));


Is topicSelector.apply(null) guaranteed to work?
Is this even the right thing to do? TopicSelector could return different topics coming from different inputs.
I think we should instead check if TopicSelector is also a LineageProvider and ask it directly.
Our TopicSelectors should then implement it and we should add to the javadoc that LineageProvider is encouraged.

Thanks for pointing this. It works only for the scenario KafkaRecordSerializationSchema.builder().setTopic(DEFAULT_TOPIC). I've changed the implementation to more clear on this case.

TopicSelector deserves more abstraction than being just Function<? super IN, String>, but I don't think this should be part of the scope of this PR.

AHeise · 2024-10-15T09:16:36Z

...c/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java

+            facets.add(new KafkaTopicListFacet(Arrays.asList(topicSelector.apply(null))));
+
+            // gets type information from serialize method signature
+            Arrays.stream(this.valueSerializationSchema.getClass().getMethods())


Again we should probably check for the serializer to return the TypeInformation directly (by implementing ResultTypeQueryable).
If not we could fallback to extract that as you do, but I'd use things like org.apache.flink.shaded.guava31.com.google.common.reflect.TypeToken to be more robust. Your implementation fails if you have some intermediate interface that forward the type parameter.

Checking for ResultTypeQueryable first is fair. I've switched to guava reflection helpers as suggested, but I am not sure if this helps with implementation fails if you have some intermediate interface that forward the type parameter.. Could you provide example of this issue? Not sure if this is now covered or not.

AHeise · 2024-10-15T09:42:04Z

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

+        facets.addAll(LineageUtil.facetsFrom(recordSerializer));
+        facets.add(new KafkaPropertiesFacet(this.kafkaProducerConfig));


See feedback on LineageUtil

AHeise · 2024-10-24T18:28:59Z

...r-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/facets/KafkaDatasetFacet.java

+            this.topicPattern = topicPattern;
+        }
+
+        public static KafkaDatasetIdentifier of(Pattern pattern) {


Suggested change

public static KafkaDatasetIdentifier of(Pattern pattern) {

public static KafkaDatasetIdentifier ofPattern(Pattern pattern) {

AHeise

This is definitively heading into the right direction. Structure of production code is good as-is but the handling of the properties doesn't look fully right.

Tests look complete at the first class but we need to get rid of mockito :).

AHeise · 2024-10-24T18:29:21Z

...r-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/facets/KafkaDatasetFacet.java

+            return new KafkaDatasetIdentifier(Collections.emptyList(), pattern);
+        }
+
+        public static KafkaDatasetIdentifier of(List<String> fixedTopics) {


Suggested change

public static KafkaDatasetIdentifier of(List<String> fixedTopics) {

public static KafkaDatasetIdentifier ofTopics(List<String> fixedTopics) {

AHeise · 2024-10-24T18:29:43Z

...r-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/facets/KafkaDatasetFacet.java

+     * Record class to contain topics' identifier information which can be either a list of topics
+     * or a topic pattern.
+     */
+    public static class KafkaDatasetIdentifier {


Any reason to not make it top-level?

AHeise · 2024-10-24T18:30:41Z

...r-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/facets/KafkaDatasetFacet.java

@@ -0,0 +1,97 @@
+package org.apache.flink.connector.kafka.lineage.facets;


Is it coming to have a separate package for facets? If not, I'd use a package for all lineage classes. There are not that many.

AHeise · 2024-10-24T18:31:48Z

...r-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/facets/KafkaDatasetFacet.java

+
+    public static final String KAFKA_FACET_NAME = "kafka";
+
+    public final Properties properties;


We usually avoid public and rather use the full jazz. It just makes it easier to later add more validation or defensive copies when needed.

AHeise · 2024-10-24T18:32:12Z

...r-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/facets/KafkaDatasetFacet.java

+        this.typeInformation = typeInformation;
+    }
+
+    public void addProperties(Properties properties) {


Since this method modifies the properties, the ctor should make a copy.

AHeise · 2024-10-24T20:29:38Z

...c/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java

+                        Arrays.stream(this.valueSerializationSchema.getClass().getMethods())
+                                .map(m -> Invokable.from(m))
+                                .filter(m -> "serialize".equalsIgnoreCase(m.getName()))
+                                .map(m -> m.getParameters().get(0))
+                                .filter(p -> !p.getType().equals(TypeToken.of(Object.class)))
+                                .findFirst()
+                                .map(p -> p.getType())
+                                .map(t -> TypeInformation.of(t.getRawType()))
+                                .orElse(null);


This looks way more complicated as it should be. Here is what I had in mind.

TypeToken<? extends SerializationSchema> serializationSchemaType = TypeToken.of(valueSerializationSchema.getClass()); Class<?> parameterType = serializationSchemaType.resolveType(SerializationSchema.class.getTypeParameters()[0]).getRawType(); if (parameterType != Object.class) { typeInformation = TypeInformation.of(parameterType); }

AHeise · 2024-10-24T20:32:12Z

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

+
+            if (!kafkaDatasetFacet.isPresent()) {
+                LOG.warn("Provided did not return kafka dataset facet");
+                return null;


I don't think we are allowed to return null. The interface doesn't specify @nullable.

Nope, interface doesn't specify @nullable. We can return LineageVertex with empty dataset list instead.

AHeise · 2024-10-24T20:35:07Z

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

+                return null;
+            }
+
+            kafkaDatasetFacet.get().addProperties(this.kafkaProducerConfig);


Do we ever actually get the properties from the recordSerializer? So are we actually just setting here?

ok, we can convert it to setter.

AHeise · 2024-10-24T20:37:11Z

...c/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java

+                        ((ResultTypeQueryable<?>) this.valueSerializationSchema).getProducedType();
+            } else {
+                // gets type information from serialize method signature
+                typeInformation =


How do we use this type information later? This is the input type, right?

This is returned within the facet and then listener (like OpenLineageJobListener) converts it to dataset schema format description. For OpenLineage, it's called SchemaDatasetFacet. I think this is not Kafka connector specific and there should be a general schema-alike facet within flink core. However, I don't feel I would be able to achieve this now. Schema information is valuable for both input and output datasets.

I hope typeInformation approach will work well for Avro and Protobuf. Hopefully, in some time, I create separate tests within OpenLineage job listener to verify this.

Yes TypeInformationFacet sounds like a general concept. I'm convinced you want to pull it out of the KafkaFacet now. You probably want to name it "inputType" and "outputType" depending on the type of the connector (source/sink). I'd design it generally and pull it up into flink-core for Flink 2.0 later (so make it work in Kafka first and then propose to port it upwards).

LineageGraph in the flink-core contains separate lists of sources and sinks. Given that, I am not sure if we want to distinguish "inputType" from "outputType". From the facet perspective, this should be all type and the same facet can be used for both scenarios.

AHeise · 2024-10-24T20:41:16Z

...st/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java

+        when(((KafkaDatasetIdentifierProvider) topicSelector).getDatasetIdentifier())
+                .thenReturn(Optional.empty());


I haven't looked too closely at the tests. But a high-level comment: In Flink, we don't use mockito (anymore). The main idea is that we use interfaces (as you did) and then just explicitly create our MockImplementation.

class MockTopicSelector implements TopicSelector, KafkaDatasetIdentifierProvider { KafkaDatasetIdentifier id; // init with ctor or factory method KafkaDatasetIdentifier getDatasetIdentifier() { return id; } }

HuangZhenQiu · 2024-10-29T18:37:53Z

...-connector-kafka/src/test/java/org/apache/flink/connector/kafka/lineage/LineageUtilTest.java

+public class LineageUtilTest {
+    @Test
+    public void testSourceLineageVertexOf() {
+        LineageDataset dataset = Mockito.mock(LineageDataset.class);


As called out by @AHeise, we need to move out from Mockito with testing classes. I am thinking. I should probably add these helper test classes in flink-core rather than implement in each of connector.

Thanks @HuangZhenQiu for noticing that place.

Signed-off-by: Pawel Leszczynski <leszczynski.pawel@gmail.com>

HuangZhenQiu

Thanks for the contribution. This diff give a great example for connectors to support flink native lineage.

AHeise

LAGTM. A few more nits. The most important part is around documentation. Make sure all Public elements are properly annotated and that you link from existing interfaces to the new optional mixins.

AHeise · 2024-11-05T07:31:38Z

...ka/src/main/java/org/apache/flink/connector/kafka/lineage/DefaultKafkaDatasetIdentifier.java

+    @Nullable private final List<String> topics;
+    @Nullable private final Pattern topicPattern;
+
+    public DefaultKafkaDatasetIdentifier(List<String> fixedTopics, Pattern topicPattern) {


Suggested change

public DefaultKafkaDatasetIdentifier(List<String> fixedTopics, Pattern topicPattern) {

public DefaultKafkaDatasetIdentifier(@Nullable List<String> fixedTopics, @Nullable Pattern topicPattern) {

Just try to be as consistent as possible.

AHeise · 2024-11-11T09:26:07Z

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/LineageUtil.java

+        if (bootstrapServers.contains(COMMA)) {
+            bootstrapServers = bootstrapServers.split(COMMA)[0];
+        } else if (bootstrapServers.contains(SEMICOLON)) {
+            bootstrapServers = bootstrapServers.split(SEMICOLON)[0];
+        }


Can you check if there is already some util in kafka that does that? If not, leave as is.

Seems like piece of code that has to be available somewhere, but I wasn't able to find it.

AHeise · 2024-11-11T09:27:18Z

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/LineageUtil.java

+
+            @Override
+            public List<LineageDataset> datasets() {
+                return datasets.stream().collect(Collectors.toList());


Suggested change

return datasets.stream().collect(Collectors.toList());

return List.copyOf(datasets);

AHeise · 2024-11-11T09:27:50Z

...r-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/TypeDatasetFacetProvider.java

+     * Returns a type dataset facet or `Optional.empty` in case an implementing class is not able to
+     * resolve type.
+     *
+     * @return


Please remove all empty javadoc tags or let Copilot help you ;)

AHeise · 2024-11-11T09:30:29Z

...a/src/main/java/org/apache/flink/connector/kafka/lineage/KafkaDatasetIdentifierProvider.java

+import java.util.Optional;
+
+/** Contains method which allows extracting topic identifier. */
+public interface KafkaDatasetIdentifierProvider {


Make sure to tag all public API with @PublicEvolving. It needs to be clearly visible if a user is supposed to touch the class or not (the easiest way is to not use public unless needed).

AHeise · 2024-11-11T09:32:47Z

...va/org/apache/flink/connector/kafka/source/enumerator/subscriber/PartitionSetSubscriber.java

 import java.util.Set;
 import java.util.stream.Collectors;

 import static org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaSubscriberUtils.getTopicMetadata;

 /** A subscriber for a partition set. */
-class PartitionSetSubscriber implements KafkaSubscriber {
+class PartitionSetSubscriber implements KafkaDatasetIdentifierProvider, KafkaSubscriber {


Suggested change

class PartitionSetSubscriber implements KafkaDatasetIdentifierProvider, KafkaSubscriber {

class PartitionSetSubscriber implements KafkaSubscriber, KafkaDatasetIdentifierProvider {

keep it consistent

AHeise · 2024-11-11T09:37:45Z

...st/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java

+                        .setKeySerializationSchema(serializationSchema)
+                        .build();
+
+        assertThat(((KafkaDatasetFacetProvider) schema).getKafkaDatasetFacet()).isEmpty();


A bit more assertj-ish would be

assertThat(schema) .asInstanceOf(InstanceOfAssertFactories.type(KafkaDatasetFacetProvider.class)) .returns(List.of(), KafkaDatasetFacetProvider::getKafkaDatasetFacet);

That would result in an assertion error instead of runtime error if the Schema does not implement the interface.

AHeise · 2024-11-11T09:39:28Z

...c/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java

@@ -79,6 +95,7 @@
 */
 @PublicEvolving
 public class KafkaRecordSerializationSchemaBuilder<IN> {
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaSource.class);


Not the correct place, but please update the docs of the KafkaRecordSerializationSchema to point to the FacetProvider interface. Same to all other APIs where you hope that optional interfaces are implemented.

AHeise · 2024-11-11T09:41:02Z

...k-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceTest.java

+        KafkaSource source =
+                new KafkaSource(
+                        new KafkaSubscriber() {
+                            @Override
+                            public Set<TopicPartition> getSubscribedTopicPartitions(
+                                    AdminClient adminClient) {
+                                return null;
+                            }
+                        },
+                        null,
+                        null,
+                        Boundedness.CONTINUOUS_UNBOUNDED,
+                        null,
+                        kafkaProperties,
+                        null);
+        assertThat(source.getLineageVertex().datasets()).isEmpty();


Can you use the builder instead? That should also be less verbose.

AHeise

LGTM. I triggered the hopefully final CI run.

Signed-off-by: Pawel Leszczynski <leszczynski.pawel@gmail.com>

boring-cyborg · 2024-11-14T08:40:21Z

Awesome work, congrats on your first merged pull request!

AHeise · 2024-11-14T08:40:38Z

Thank you very much for your contribution (and patience).

pawel-big-lebowski · 2024-11-14T09:39:32Z

Awesome. @AHeise Thank you for your feedback and cooperation on that.

davidradl · 2024-11-15T10:19:22Z

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/lineage/LineageUtil.java

+        return new LineageVertex() {
+            @Override
+            public List<LineageDataset> datasets() {
+                return datasets.stream().collect(Collectors.toList());


Can we not use the List.copyOf(datasets); as per Arvid's suggested change here also.

Merged version is not using List.copyOf. We agreed on that in offline discussion.

davidradl · 2024-11-15T11:55:16Z

...c/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java

+                TypeToken serializationSchemaType =
+                        TypeToken.of(valueSerializationSchema.getClass());
+                Class parameterType =
+                        serializationSchemaType


Is there a way to avoid using this reflection - instanceof and Class. Maybe using a config driven approach and java SPI. The connectors / formats bring in serialization implementations in this way that avoid the overhead of reflection.

You can avoid this with implementing ResultTypeQueryable of the value serialisation schema.

boring-cyborg bot added the component=Connectors/Kafka label Oct 14, 2024

pawel-big-lebowski force-pushed the lineage-impl branch 2 times, most recently from 6bf144b to 5834583 Compare October 14, 2024 12:55

HuangZhenQiu reviewed Oct 15, 2024

View reviewed changes

pawel-big-lebowski changed the title ~~Lineage interfaces for kafka connector~~ [FLINK-34466] Lineage interfaces for kafka connector Oct 15, 2024

AHeise reviewed Oct 15, 2024

View reviewed changes

pawel-big-lebowski force-pushed the lineage-impl branch 2 times, most recently from 6d7b4a0 to baafb00 Compare October 16, 2024 11:47

[FLINK-34466] lineage interfaces for kafka connector

ca14634

pawel-big-lebowski force-pushed the lineage-impl branch from baafb00 to ca14634 Compare October 16, 2024 12:09

pawel-big-lebowski requested a review from AHeise October 16, 2024 12:29

pawel-big-lebowski force-pushed the lineage-impl branch from 0484abd to ea71a84 Compare October 18, 2024 10:10

AHeise reviewed Oct 24, 2024

View reviewed changes

pawel-big-lebowski force-pushed the lineage-impl branch 2 times, most recently from b25b5b7 to 27d8903 Compare October 29, 2024 13:52

HuangZhenQiu reviewed Oct 29, 2024

View reviewed changes

[FLINK-34466] create KafkaDatasetFacet

4bbff17

Signed-off-by: Pawel Leszczynski <leszczynski.pawel@gmail.com>

pawel-big-lebowski force-pushed the lineage-impl branch from 27d8903 to 4bbff17 Compare October 30, 2024 07:13

HuangZhenQiu approved these changes Oct 31, 2024

View reviewed changes

AHeise reviewed Nov 11, 2024

View reviewed changes

pawel-big-lebowski requested a review from AHeise November 12, 2024 12:48

AHeise approved these changes Nov 13, 2024

View reviewed changes

code review changes

4f7acc0

Signed-off-by: Pawel Leszczynski <leszczynski.pawel@gmail.com>

pawel-big-lebowski force-pushed the lineage-impl branch from e751699 to 4f7acc0 Compare November 13, 2024 13:35

AHeise merged commit 727327d into apache:main Nov 14, 2024
6 checks passed

davidradl reviewed Nov 15, 2024

View reviewed changes

		facets.addAll(LineageUtil.facetsFrom(recordSerializer));
		facets.add(new KafkaPropertiesFacet(this.kafkaProducerConfig));

	public static KafkaDatasetIdentifier of(Pattern pattern) {
	public static KafkaDatasetIdentifier ofPattern(Pattern pattern) {

	public static KafkaDatasetIdentifier of(List<String> fixedTopics) {
	public static KafkaDatasetIdentifier ofTopics(List<String> fixedTopics) {

		@@ -0,0 +1,97 @@
		package org.apache.flink.connector.kafka.lineage.facets;


		public static final String KAFKA_FACET_NAME = "kafka";

		public final Properties properties;

		when(((KafkaDatasetIdentifierProvider) topicSelector).getDatasetIdentifier())
		.thenReturn(Optional.empty());

	public DefaultKafkaDatasetIdentifier(List<String> fixedTopics, Pattern topicPattern) {
	public DefaultKafkaDatasetIdentifier(@Nullable List<String> fixedTopics, @Nullable Pattern topicPattern) {

	return datasets.stream().collect(Collectors.toList());
	return List.copyOf(datasets);

	class PartitionSetSubscriber implements KafkaDatasetIdentifierProvider, KafkaSubscriber {
	class PartitionSetSubscriber implements KafkaSubscriber, KafkaDatasetIdentifierProvider {

[FLINK-34466] Lineage interfaces for kafka connector #130

[FLINK-34466] Lineage interfaces for kafka connector #130

Conversation

pawel-big-lebowski commented Oct 14, 2024 • edited Loading

boring-cyborg bot commented Oct 14, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HuangZhenQiu commented Oct 15, 2024 • edited Loading

AHeise left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AHeise left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AHeise Oct 24, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HuangZhenQiu left a comment

Choose a reason for hiding this comment

AHeise left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AHeise left a comment

Choose a reason for hiding this comment

boring-cyborg bot commented Nov 14, 2024

AHeise commented Nov 14, 2024

pawel-big-lebowski commented Nov 14, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pawel-big-lebowski commented Oct 14, 2024 •

edited

Loading

HuangZhenQiu commented Oct 15, 2024 •

edited

Loading

AHeise Oct 24, 2024 •

edited

Loading