I am currently running a small application that periodically polls data from my DB and then puts it in a Kafka topic. While running the application code independently, when I comment my Kafka sink, the application runs correctly:
KafkaSink<String> stringSink = KafkaSinkFactory.createStringKafkaSink(props, kafkaProps, props.getProperty("asset.sync.kafka.sink.topic.name")); dbStream .map(Object::toString) // or map your custom object to String .sinkTo(stringSink) .name("String Kafka Sink"); But running both the sink and the source together gives the exception:
java.sql.SQLRecoverableException: IO Error: Socket read interrupted, Authentication lapse 0 ms. at oracle.jdbc.driver.T4CConnection.logon(T4CConnection.java:874) at oracle.jdbc.driver.PhysicalConnection.connect(PhysicalConnection.java:793) at oracle.jdbc.driver.T4CDriverExtension.getConnection(T4CDriverExtension.java:57) at oracle.jdbc.driver.OracleDriver.connect(OracleDriver.java:747) at oracle.jdbc.driver.OracleDriver.connect(OracleDriver.java:562) at com.zaxxer.hikari.util.DriverDataSource.getConnection(DriverDataSource.java:138) at com.zaxxer.hikari.pool.PoolBase.newConnection(PoolBase.java:364) at com.zaxxer.hikari.pool.PoolBase.newPoolEntry(PoolBase.java:206) at com.zaxxer.hikari.pool.HikariPool.createPoolEntry(HikariPool.java:476) at com.zaxxer.hikari.pool.HikariPool.checkFailFast(HikariPool.java:561) at com.zaxxer.hikari.pool.HikariPool.<init>(HikariPool.java:115) at com.zaxxer.hikari.HikariDataSource.<init>(HikariDataSource.java:81) at ***.***.hcmp.config.DBConfig.getDataSource(DBConfig.java:22) at ***.***.hcmp.source.assetMaster.DBPollingSource.open(DBPollingSource.java:62) at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:34) at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:101) at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:107) at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:753) at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.call(StreamTaskActionExecutor.java:100) at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:728) at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:693) at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:953) at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:922) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) at java.base/java.lang.Thread.run(Thread.java:829) Caused by: java.io.IOException: Socket read interrupted, Authentication lapse 0 ms. at oracle.jdbc.driver.T4CConnection.logon(T4CConnection.java:870) ... 25 common frames omitted Caused by: java.io.InterruptedIOException: Socket read interrupted at oracle.net.nt.TimeoutSocketChannel.handleInterrupt(TimeoutSocketChannel.java:258) at oracle.net.nt.TimeoutSocketChannel.read(TimeoutSocketChannel.java:180) at oracle.net.ns.NSProtocolNIO.doSocketRead(NSProtocolNIO.java:555) at oracle.net.ns.NIOPacket.readNIOPacket(NIOPacket.java:403) at oracle.net.ns.NSProtocolNIO.negotiateConnection(NSProtocolNIO.java:127) at oracle.net.ns.NSProtocol.connect(NSProtocol.java:340) at oracle.jdbc.driver.T4CConnection.connect(T4CConnection.java:1596) at oracle.jdbc.driver.T4CConnection.logon(T4CConnection.java:588) Here is the application code:
public class RefreshPollingApplication { private static final Logger log = LoggerFactory.getLogger(RefreshPollingApplication.class); public static void main(String[] args) throws Exception { ParameterTool params = ParameterTool.fromPropertiesFile( RefreshPollingApplication.class.getClassLoader().getResourceAsStream("application-local.properties") ); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.getConfig().setGlobalJobParameters(params); env.setParallelism(1); env.setRestartStrategy(RestartStrategies.fixedDelayRestart( 3, Time.of(10, TimeUnit.SECONDS) )); try{ Properties props = ConfigLoader.load("application-local.properties"); Properties kafkaProps = getKafkaProperties(props); log.info("@@@@@@@@@@@@@ Properties loaded"); SimpleDBConfig dbConfig = new SimpleDBConfig(params.get("flink.jdbc.url"), params.get("flink.jdbc.username"), params.get("flink.jdbc.password")); Connection connection = dbConfig.getConnection(); DataStream<String> dbStream = env.addSource( new JdbcPollingSource() ).setParallelism(1).name("DB Polling Source"); KafkaSink<String> stringSink = KafkaSinkFactory.createStringKafkaSink(props, kafkaProps, props.getProperty("****.***.kafka.sink.topic.name")); dbStream .map(Object::toString) .sinkTo(stringSink) .name("String Kafka Sink"); env.execute("Refresh polling job."); }catch (Exception e){ log.error("@@@@@@@@ERROR OCCURRED ON POLLING TASK, REASON: {}", e.getMessage()); throw e; } } } Source function class:
public class JdbcPollingSource extends RichSourceFunction<String> { private volatile boolean isRunning = true; private transient Connection connection; private transient PreparedStatement stmt; private String url; private String userName; private String password; @Override public void open(Configuration parameters) throws Exception { Class.forName("oracle.jdbc.OracleDriver"); super.open(parameters); Map<String, String> globalParams = getRuntimeContext() .getExecutionConfig() .getGlobalJobParameters() .toMap(); ParameterTool params = ParameterTool.fromMap(globalParams); this.url = params.get("flink.jdbc.url"); userName = params.get("flink.jdbc.username"); password = params.get("flink.jdbc.password"); reconnect(); } private void reconnect() throws Exception { if (connection != null && !connection.isClosed()) { try { connection.close(); } catch (Exception ignored) {} } try { connection = DriverManager.getConnection( url, userName, password ); stmt = connection.prepareStatement( "*******" ); }catch (Exception e){ System.out.println("Error connecting to data source, exception -> " + e.getMessage()); } } @Override public void run(SourceContext<String> ctx) throws Exception { while (isRunning) { try { ResultSet rs = stmt.executeQuery(); while (rs.next()) { String record = rs.getLong("***") + "," + rs.getString("***"); ctx.collect(record); } } catch (SQLException e) { reconnect(); } Thread.sleep(120_000); } } @Override public void cancel() { isRunning = false; try { if (stmt != null) stmt.close(); } catch (Exception ignore) {} try { if (connection != null) connection.close(); } catch (Exception ignore) {} } } Is it possible that a background thread is interfering with the DB connection here?
