snuyanzin commented on code in PR #235:
URL:
https://github.com/apache/flink-connector-jdbc/pull/235#discussion_r4065795364
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSource.java:
##########
@@ -120,16 +124,27 @@ public Boundedness getBoundedness() {
public SourceReader<OUT, JdbcSourceSplit> createReader(SourceReaderContext
readerContext)
throws Exception {
return new JdbcSourceReader<>(
- () ->
- new JdbcSourceSplitReader<>(
- readerContext,
- configuration,
- typeInformation,
- connectionProvider,
- deliveryGuarantee,
- resultExtractor),
- configuration,
- readerContext);
+ splitReaderSupplier(readerContext), configuration,
readerContext);
+ }
+
+ /**
+ * Supplies the split readers of one {@link JdbcSourceReader}. Every
reader gets its own
+ * connection provider: a provider holds a single connection and is not
thread safe, while
+ * flink-connector-base builds a split reader per split fetcher and closes
the reader of a
+ * finished split while the reader of the next split is already reading.
Sharing one provider
+ * lets the first close the connection out from under the second.
+ */
+ @VisibleForTesting
+ Supplier<SplitReader<RecordAndOffset<OUT>, JdbcSourceSplit>>
splitReaderSupplier(
Review Comment:
seems we still use old ArchUnit here...
newer (1.4.x ) would require regeneration after adding annotation like 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
