Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Fix-637][efak-plugin] Fixed mysql plugin auto create databases and tables #668

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[Fix-637][efak-plugin] Fixed mysql plugin auto create databases and tables #668

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion efak-common/src/main/resources/system-config.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,7 @@ cluster1.efak.ssl.cgroup.topics=
# kafka jdbc driver address
######################################
efak.driver=com.mysql.cj.jdbc.Driver
efak.url=jdbc:mysql://127.0.0.1:3306/ke_prd?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
efak.url=jdbc:mysql://127.0.0.1:3306/ke_prd?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&serverTimezone=Asia/Shanghai
efak.username=root
efak.password=123456
#efak.driver=org.sqlite.JDBC
Expand Down
20 changes: 12 additions & 8 deletions efak-plugin/src/main/java/org/smartloli/kafka/eagle/plugin/mysql/MySqlRecordSchema.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,17 +47,21 @@ public static void schema() {
String url = SystemConfigUtils.getProperty("efak.url");
String username = SystemConfigUtils.getProperty("efak.username");
String password = SystemConfigUtils.getProperty("efak.password");
String host = url.split("//")[1].split("/")[0].split(":")[0];
String port = url.split("//")[1].split("/")[0].split(":")[1];
String db = url.split("//")[1].split("/")[1].split("\\?")[0];
int prefixLength = url.split("//")[0].length()
+ url.split("//")[1].split("/")[0].length() + "?".length()
+ 3;
int dbNameLength = db.length();
// exclude db: "jdbc:mysql://127.0.0.1:3306" + "?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull"
String databaseUrl = url.substring(0, prefixLength - 2) + url.substring(prefixLength + dbNameLength - 1);

if (database(username, password, host, port, db)) {
tables(username, password, host, port, db);
if (database(username, password, databaseUrl, db)) {
tables(username, password, url);
}
}

private static void tables(String username, String password, String host, String port, String db) {
Connection connection = MySqlStoragePlugin.getInstance(host + ":" + port + "/" + db, username, password);
private static void tables(String username, String password, String url) {
Connection connection = MySqlStoragePlugin.getInstance(url, username, password);
ResultSet rs = null;
Statement stmt = null;
List<String> tbls = new ArrayList<>();
Expand Down Expand Up @@ -125,8 +129,8 @@ private static void tables(String username, String password, String host, String
MySqlStoragePlugin.close(connection);
}

private static boolean database(String username, String password, String host, String port, String db) {
Connection connection = MySqlStoragePlugin.getInstance(host + ":" + port, username, password);
private static boolean database(String username, String password, String databaseUrl, String db) {
Connection connection = MySqlStoragePlugin.getInstance(databaseUrl, username, password);
ResultSet rs = null;
List<String> dbs = new ArrayList<>();
try {
Expand Down
2 changes: 1 addition & 1 deletion efak-plugin/src/main/java/org/smartloli/kafka/eagle/plugin/mysql/MySqlStoragePlugin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ public class MySqlStoragePlugin {
public static Connection getInstance(String address, String username, String password) {
Connection connection = null;
try {
connection = (Connection) DriverManager.getConnection("jdbc:mysql://" + address, username, password);
connection = (Connection) DriverManager.getConnection(address, username, password);
} catch (Exception e) {
LOG.error("Create mysql connection has error address[" + address + "],username[" + username + "],password[" + password + "],msg is " + e.getMessage());
}
Expand Down
2 changes: 1 addition & 1 deletion efak-web/src/main/resources/conf/system-config.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ cluster3.efak.ssl.cgroup.topics=
# kafka mysql jdbc driver address
######################################
efak.driver=com.mysql.cj.jdbc.Driver
efak.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
efak.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&serverTimezone=Asia/Shanghai
efak.username=root
efak.password=123456