featzhang commented on code in PR #123:
URL:
https://github.com/apache/flink-connector-pulsar/pull/123#discussion_r3257462233
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceFetcherManager.java:
##########
@@ -59,16 +56,13 @@ public class PulsarSourceFetcherManager
/**
* Creates a new SplitFetcherManager with multiple I/O threads.
*
- * @param elementsQueue The queue that is used to hand over data from the
I/O thread (the
- * fetchers) to the reader, which emits the records and book-keeps the
state. This must be
- * the same queue instance that is also passed to the {@link
SourceReaderBase}.
* @param splitReaderSupplier The factory for the split reader that
connects to the source
+ * @param configuration The configuration for the fetcher manager
*/
public PulsarSourceFetcherManager(
-
FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>>
elementsQueue,
Supplier<SplitReader<Message<byte[]>, PulsarPartitionSplit>>
splitReaderSupplier,
Configuration configuration) {
Review Comment:
Constructor signature change is correct for Flink 2.x. Consider adding
Javadoc to note the `elementsQueue` parameter removal.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeSerializer.java:
##########
@@ -202,7 +202,7 @@ public TypeSerializer<T> restoreSerializer() {
@Override
public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer) {
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
Review Comment:
Always returning `compatibleAsIs()` may be too permissive. Document the
serializer compatibility assumptions or add validation logic.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
