patrashu commented on code in PR #234: URL: https://github.com/apache/flink-connector-jdbc/pull/234#discussion_r4139027191
########## flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/AbstractDynamicSplitterEnumerator.java: ########## @@ -0,0 +1,124 @@ + /* + * 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.jdbc.core.datastream.source.enumerator.splitter; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.connector.source.Boundedness; +import org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit; +import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider; + +import java.io.Serializable; +import java.sql.Connection; +import java.util.Collections; +import java.util.List; + +/** + * Base class for {@link SplitterEnumerator} implementations that must open a real database + * connection at job runtime to discover their splits (e.g. by executing a user-provided query), as + * opposed to computing splits purely from static configuration. + * + * <p>Split discovery is deliberately deferred to the first call of {@link #enumerateSplits()} + * rather than performed in {@link #start(JdbcConnectionProvider)}: the enclosing {@code + * JdbcSourceEnumerator} calls {@code start()} synchronously on the JobManager coordinator thread, + * but invokes {@code enumerateSplits()} through Flink's own {@code + * SplitEnumeratorContext#callAsync}, which already runs off that thread. Deferring the actual I/O + * this way avoids blocking job startup and avoids having to manage any dedicated thread of our own. + * + * <p>Because {@code enumerateSplits()} can be invoked concurrently by the coordinator before the + * first call returns, discovery is guarded so it only ever runs once. + */ +@Internal +public abstract class AbstractDynamicSplitterEnumerator implements SplitterEnumerator { + + private transient JdbcConnectionProvider connectionProvider; + + private List<JdbcSourceSplit> discoveredSplits; + + private boolean finished = false; + + @Override + public Boundedness getBoundedness() { + return Boundedness.BOUNDED; + } + + @Override + public void start(JdbcConnectionProvider connectionProvider) { + // Intentionally no I/O here - see class javadoc. + this.connectionProvider = connectionProvider; + } + + @Override + public synchronized List<JdbcSourceSplit> enumerateSplits() { + if (finished) { + return Collections.emptyList(); + } + if (discoveredSplits == null) { + try { + Connection connection = connectionProvider.getOrEstablishConnection(); + discoveredSplits = discoverSplits(connection); + } catch (Exception e) { + throw new RuntimeException("Failed to discover splits for " + describeSource(), e); Review Comment: That's good catch. JdbcSourceEnumerator.onSplitsDiscovered now rethrows instead of logging. I traced this through Flink 2.1.3's SourceCoordinatorContext/ExecutorNotifier to confirm: the callback runs inside a ThrowableCatchingRunnable that routes any uncaught exception to handleUncaughtExceptionFromAsyncCall, which calls failJob(t) unless the coordinator is already closed. So rethrowing here does fail the job as FLIP-602 requires, instead of retrying forever. -- 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]
