Получение (java.sql.sqlrecoverableException: IO Ошибка: чтение сокета прерван, аутентификация, пропасть 0 мс.) При испол ⇐ JAVA

[Anonymous]

В настоящее время я запускаю небольшое приложение, которое периодически опробовает данные из моего БД, а затем помещает их в тему Кафки. Запустив код приложения независимо, когда я комментирую свою раковину Kafka, приложение работает правильно: < /p>

Код: Выделить всё

KafkaSink 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");
< /code>
Но запуск как раковина, так и источник вместе дает исключение:
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.(HikariPool.java:115) at com.zaxxer.hikari.HikariDataSource.(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)

Вот код приложения:

Код: Выделить всё

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 dbStream = env.addSource(
new JdbcPollingSource()
).setParallelism(1).name("DB Polling Source");
KafkaSink 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;
}
}

}
< /code>
Класс функции источника: < /p>
public class JdbcPollingSource extends RichSourceFunction {

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 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  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) {}
}
}

возможно ли, что фоновый поток мешает соединению DB?

Подробнее здесь: https://stackoverflow.com/questions/797 ... rrupted-au