You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "mas-chen (via GitHub)" <gi...@apache.org> on 2023/08/09 07:49:14 UTC
[GitHub] [flink-connector-kafka] mas-chen opened a new pull request, #44: [FLINK-32416] initial implementation of DynamicKafkaSource with bound…
mas-chen opened a new pull request, #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44
…ed/unbounded support and unit/integration tests
### What is the purpose of the change
[FLIP-246](https://cwiki.apache.org/confluence/x/CBn1D) describes the changes and this PR follows the design illustrated in the doc.
### Brief Change Log
* New Public API: DynamicKafkaSource, DynamicKafkaSourceBuilder, KafkaMetadataService and related classes, etc.
* Supports multi cluster read, dynamic cluster/topic read
* Supports multi cluster, dynamic metrics
* Supports StoppableKafkaEnumContextProxy to workaround and provide single threaded guarantees
* Supports unbounded and bounded mode
* Only supports configurations/deserializers defined uniformly across clusters, can be extended later on
* 3 stream subscription methods
* 1 KafkaMetadataService that supports single cluster, can expose more implementations in public API depending on discussion
* Supports metadata discovery interval as well as error handling strategy
### Verifying this change
* Unit Tests
* Integration Tests (including covering migration scenarios)
* Connector Test Framework Integration
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mas-chen (via GitHub)" <gi...@apache.org>.
mas-chen commented on PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#issuecomment-1887855854
one-pager docs are here: docs/content/docs/connectors/datastream/dynamic-kafka.md. Marked with experimental status.
Also copied this into the chinese doc directory.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#discussion_r1363867482
##########
flink-connector-kafka/src/test/resources/log4j2-test.properties:
##########
@@ -36,3 +36,12 @@ logger.zookeeper.name = org.apache.zookeeper
logger.zookeeper.level = OFF
logger.I0Itec.name = org.I0Itec
logger.I0Itec.level = OFF
+
+logger.DynamicKafkaSourceReader.name = org.apache.flink.connector.kafka.source.reader.DynamicKafkaSourceReader
+logger.DynamicKafkaSourceReader.level = OFF
+
+logger.kafkaSourceReader.name = org.apache.flink.connector.kafka.source.reader.KafkaSourceReader
+logger.kafkaSourceReader.level = OFF
+
+logger.SourceReaderBase.name = org.apache.flink.connector.base.source.reader.SourceReaderBase
+logger.SourceReaderBase.level = OFF
Review Comment:
Why are these disabled for tests?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/DynamicKafkaSourceOptions.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.source;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+import org.apache.flink.shaded.guava30.com.google.common.collect.Streams;
+
+import java.util.Properties;
+import java.util.function.Function;
+
+/**
+ * The connector options for {@link DynamicKafkaSource} that can be passed through the source
+ * properties e.g. {@link DynamicKafkaSourceBuilder#setProperties(Properties)}.
+ */
+@Internal
+public class DynamicKafkaSourceOptions {
+
+ private DynamicKafkaSourceOptions() {}
+
+ public static final ConfigOption<Long> STREAM_METADATA_DISCOVERY_INTERVAL_MS =
+ ConfigOptions.key("stream-metadata-discovery-interval-ms")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "The interval in milliseconds for the sink to discover "
+ + "the changes in stream metadata. A non-positive value disables the stream metadata discovery.");
Review Comment:
Long defaults to zero which could be seen as a positive value. Would set a default value of `-1` here.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/DynamicKafkaSource.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.source;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.connector.kafka.source.enumerator.DynamicKafkaSourceEnumState;
+import org.apache.flink.connector.kafka.source.enumerator.DynamicKafkaSourceEnumStateSerializer;
+import org.apache.flink.connector.kafka.source.enumerator.DynamicKafkaSourceEnumerator;
+import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
+import org.apache.flink.connector.kafka.source.enumerator.metadata.KafkaMetadataService;
+import org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaStreamSubscriber;
+import org.apache.flink.connector.kafka.source.reader.DynamicKafkaSourceReader;
+import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
+import org.apache.flink.connector.kafka.source.split.DynamicKafkaSourceSplit;
+import org.apache.flink.connector.kafka.source.split.DynamicKafkaSourceSplitSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+import java.util.Properties;
+
+/**
+ * Factory class for the DynamicKafkaSource components. <a
+ * href="https://cwiki.apache.org/confluence/x/CBn1D">FLIP-246: DynamicKafkaSource</a>
+ *
+ * <p>This source's key difference from {@link KafkaSource} is that it enables users to read
+ * dynamically from topics, which belong to one or more clusters.
Review Comment:
We should clarify what "dynamically" means.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/DynamicKafkaSource.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.source;
Review Comment:
Should this be in a separate package, e.g. `org.apache.flink.connector.kafka.dynamic.source`? This might be useful for differentiating classes from the regular Kafka source.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/DynamicKafkaSourceOptions.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.source;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+import org.apache.flink.shaded.guava30.com.google.common.collect.Streams;
+
+import java.util.Properties;
+import java.util.function.Function;
+
+/**
+ * The connector options for {@link DynamicKafkaSource} that can be passed through the source
+ * properties e.g. {@link DynamicKafkaSourceBuilder#setProperties(Properties)}.
+ */
+@Internal
+public class DynamicKafkaSourceOptions {
+
+ private DynamicKafkaSourceOptions() {}
+
+ public static final ConfigOption<Long> STREAM_METADATA_DISCOVERY_INTERVAL_MS =
+ ConfigOptions.key("stream-metadata-discovery-interval-ms")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "The interval in milliseconds for the sink to discover "
+ + "the changes in stream metadata. A non-positive value disables the stream metadata discovery.");
+
+ public static final ConfigOption<Integer> STREAM_METADATA_DISCOVERY_FAILURE_THRESHOLD =
+ ConfigOptions.key("stream-metadata-discovery-failure-threshold")
+ .intType()
+ .defaultValue(1)
+ .withDescription(
+ "The number of consecutive failures before letting the exception from Kafka metadata service discovery "
+ + "trigger jobmanager failure and global failover. The default is one to at least catch startup "
+ + "failures. This is only implemented for the source");
+
+ @Internal
+ public static <T> T getOption(
+ Properties props, ConfigOption<?> configOption, Function<String, T> parser) {
+ String value = props.getProperty(configOption.key());
+ return (T) (value == null ? configOption.defaultValue() : parser.apply(value));
+ }
+
+ @Internal
+ public static boolean containsOption(Properties props, ConfigOption<?> configOption) {
Review Comment:
Is this being used anywhere apart from tests?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/DynamicKafkaSource.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.source;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.connector.kafka.source.enumerator.DynamicKafkaSourceEnumState;
+import org.apache.flink.connector.kafka.source.enumerator.DynamicKafkaSourceEnumStateSerializer;
+import org.apache.flink.connector.kafka.source.enumerator.DynamicKafkaSourceEnumerator;
+import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
+import org.apache.flink.connector.kafka.source.enumerator.metadata.KafkaMetadataService;
+import org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaStreamSubscriber;
+import org.apache.flink.connector.kafka.source.reader.DynamicKafkaSourceReader;
+import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
+import org.apache.flink.connector.kafka.source.split.DynamicKafkaSourceSplit;
+import org.apache.flink.connector.kafka.source.split.DynamicKafkaSourceSplitSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+import java.util.Properties;
+
+/**
+ * Factory class for the DynamicKafkaSource components. <a
+ * href="https://cwiki.apache.org/confluence/x/CBn1D">FLIP-246: DynamicKafkaSource</a>
+ *
+ * <p>This source's key difference from {@link KafkaSource} is that it enables users to read
+ * dynamically from topics, which belong to one or more clusters.
+ *
+ * @param <T> Record type
+ */
+@PublicEvolving
+public class DynamicKafkaSource<T>
Review Comment:
FYI I see some references still in the FLIP which state `MultiClusterKafkaSource`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#issuecomment-1888852652
Awesome, thanks a lot! Merging once tests pass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#issuecomment-1894222901
Debug info:
```
2024-01-12T16:49:32.5164583Z [ERROR] Failures:
2024-01-12T16:49:32.5166818Z [ERROR] DynamicKafkaSourceITTest$IntegrationTests>SourceTestSuiteBase.testMultipleSplits:211->SourceTestSuiteBase.checkResultWithSemantic:743 Expected to have exactly 40 records in result, but only received 30 records
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mas-chen (via GitHub)" <gi...@apache.org>.
mas-chen commented on PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#issuecomment-1882369061
@tzulitai Thanks for the review. I would like get some more input from at least another reviewer from merging this and I agree this should be included in the 3.1.0 release. Perhaps @mxm?
After this PR, I still need to complete the documentation for this new connector.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#issuecomment-1754958456
Hey Mason! Do you want a review for this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mas-chen (via GitHub)" <gi...@apache.org>.
mas-chen commented on code in PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#discussion_r1448066046
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/DynamicKafkaSource.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.source;
Review Comment:
Let me know if you have any other considerations. I'll squash the commits if we are in agreement (leaving the dangling commit now for your reviewing convenience)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm merged PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mas-chen (via GitHub)" <gi...@apache.org>.
mas-chen commented on code in PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#discussion_r1447804230
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/DynamicKafkaSource.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.source;
Review Comment:
I actually prefer `dynamic.source` since it makes sense for future usecases to leverage the dynamic code. e.g. a `dynamic.sink`. With that being said, I'm going to move the metadata up one package since that isn't source specific.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mas-chen (via GitHub)" <gi...@apache.org>.
mas-chen commented on PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#issuecomment-1887602638
Are the Javadocs sufficient in the bare minimum case? Otherwise, I'll just write the documentation into this PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mas-chen (via GitHub)" <gi...@apache.org>.
mas-chen commented on PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#issuecomment-1887586297
Thanks @tzulitai @mxm for your review! I've addressed all the comments and squashed the commits. Can you please merge it at your convenience?
I'll start on the docs immediately and hope to use this connector directly in the 3.1.0 release!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mas-chen (via GitHub)" <gi...@apache.org>.
mas-chen commented on code in PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#discussion_r1445596890
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/DynamicKafkaSource.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.source;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.connector.kafka.source.enumerator.DynamicKafkaSourceEnumState;
+import org.apache.flink.connector.kafka.source.enumerator.DynamicKafkaSourceEnumStateSerializer;
+import org.apache.flink.connector.kafka.source.enumerator.DynamicKafkaSourceEnumerator;
+import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
+import org.apache.flink.connector.kafka.source.enumerator.metadata.KafkaMetadataService;
+import org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaStreamSubscriber;
+import org.apache.flink.connector.kafka.source.reader.DynamicKafkaSourceReader;
+import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
+import org.apache.flink.connector.kafka.source.split.DynamicKafkaSourceSplit;
+import org.apache.flink.connector.kafka.source.split.DynamicKafkaSourceSplitSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+import java.util.Properties;
+
+/**
+ * Factory class for the DynamicKafkaSource components. <a
+ * href="https://cwiki.apache.org/confluence/x/CBn1D">FLIP-246: DynamicKafkaSource</a>
+ *
+ * <p>This source's key difference from {@link KafkaSource} is that it enables users to read
+ * dynamically from topics, which belong to one or more clusters.
+ *
+ * @param <T> Record type
+ */
+@PublicEvolving
+public class DynamicKafkaSource<T>
Review Comment:
Yes, I was a bit lazy there, since it required the diagrams to change. Let me revisit that now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mas-chen (via GitHub)" <gi...@apache.org>.
mas-chen commented on PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#issuecomment-1758623662
@mxm Yes please! I've rebased the PR and working on fixing some flaky unit tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mas-chen (via GitHub)" <gi...@apache.org>.
mas-chen commented on PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#issuecomment-1895038870
Sorry about that. ^The linked PR should fix this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#discussion_r1448982380
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/DynamicKafkaSource.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.source;
Review Comment:
Good to go. I can squash also on merge, as you prefer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#issuecomment-1887599176
Sounds good. Can we get a bare-minimum version of the documentation into this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mas-chen (via GitHub)" <gi...@apache.org>.
mas-chen commented on code in PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#discussion_r1445538272
##########
flink-connector-kafka/src/test/resources/log4j2-test.properties:
##########
@@ -36,3 +36,12 @@ logger.zookeeper.name = org.apache.zookeeper
logger.zookeeper.level = OFF
logger.I0Itec.name = org.I0Itec
logger.I0Itec.level = OFF
+
+logger.DynamicKafkaSourceReader.name = org.apache.flink.connector.kafka.source.reader.DynamicKafkaSourceReader
+logger.DynamicKafkaSourceReader.level = OFF
+
+logger.kafkaSourceReader.name = org.apache.flink.connector.kafka.source.reader.KafkaSourceReader
+logger.kafkaSourceReader.level = OFF
+
+logger.SourceReaderBase.name = org.apache.flink.connector.base.source.reader.SourceReaderBase
+logger.SourceReaderBase.level = OFF
Review Comment:
This was for troubleshooting integration tests. I'll remove them now that they are stable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#issuecomment-1887622480
A brief one-pager of the feature would be good. We can expand on it later. Just to have something in the web docs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#issuecomment-1894135321
NB: after this was merged, the CI doesn't work for `main` anymore CC @mas-chen @mxm
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#discussion_r1447193636
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/DynamicKafkaSource.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.source;
Review Comment:
On hindsight, I wonder whether `source.dynamic` makes more sense than `dynamic.source` 🙈
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#issuecomment-1894220601
We also noticed that. @mas-chen is looking into fixing this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-32416] initial implementation of DynamicKafkaSource with bound… [flink-connector-kafka]
Posted by "mas-chen (via GitHub)" <gi...@apache.org>.
mas-chen commented on code in PR #44:
URL: https://github.com/apache/flink-connector-kafka/pull/44#discussion_r1446686787
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumState.java:
##########
@@ -42,7 +42,7 @@ public KafkaSourceEnumState(
this.initialDiscoveryFinished = initialDiscoveryFinished;
}
- KafkaSourceEnumState(
+ public KafkaSourceEnumState(
Review Comment:
This is the only KafkaSource code that I had to modify. As a result of moving packages, this needed to be publicly visible (the class is still annotated as Internal)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org