Mobile versionCode: Select all
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>
Aber das Ausführen sowohl der Spüle als auch der Quelle ergibt die Ausnahme:
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)Hier ist der Anwendungscode:
Code: Select all
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>
Quellfunktionsklasse: < /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) {}
}
}
