style: fix ide warnings (#17)

flink-connector-oceanbase/src/main/java/com/oceanbase/connector/flink/connection/OceanBaseConnectionPool.java

@@ -33,7 +33,7 @@ public OceanBaseConnectionPool(OceanBaseConnectionOptions options) {
  this.options = options;
  }
 
- public void init() throws SQLException {
+ public void init() {
  if (!inited) {
  synchronized (this) {
  if (!inited) {

flink-connector-oceanbase/src/main/java/com/oceanbase/connector/flink/connection/OceanBaseConnectionProvider.java

@@ -15,6 +15,7 @@
 import java.sql.Connection;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Statement;
 
 public interface OceanBaseConnectionProvider extends AutoCloseable {
 
@@ -33,11 +34,10 @@ public interface OceanBaseConnectionProvider extends AutoCloseable {
  * @throws SQLException if a database access error occurs
  */
  default String getCompatibleMode() throws SQLException {
- try (Connection conn = getConnection()) {
- ResultSet rs =
- conn.createStatement()
- .executeQuery("SHOW VARIABLES LIKE 'ob_compatibility_mode'");
- while (rs.next()) {
+ try (Connection conn = getConnection();
+ Statement statement = conn.createStatement()) {
+ ResultSet rs = statement.executeQuery("SHOW VARIABLES LIKE 'ob_compatibility_mode'");
+ if (rs.next()) {
  return rs.getString("Value");
  }
  return null;

flink-connector-oceanbase/src/main/java/com/oceanbase/connector/flink/sink/OceanBaseRowDataStatementExecutor.java

@@ -199,8 +199,8 @@ private boolean exist(RowData rowData) throws SQLException {
  if (closed || rowData == null) {
  return true;
  }
- try (Connection connection = connectionProvider.getConnection()) {
- PreparedStatement statement = connection.prepareStatement(existStatementSql);
+ try (Connection connection = connectionProvider.getConnection();
+  PreparedStatement statement = connection.prepareStatement(existStatementSql)) {
  setStatementData(statement, rowData, tableSchema.getKeyFieldGetters());
  ResultSet resultSet = statement.executeQuery();
  return resultSet.next();
@@ -213,8 +213,8 @@ private void executeBatch(
  if (closed || rowDataList == null || rowDataList.isEmpty()) {
  return;
  }
- try (Connection connection = connectionProvider.getConnection()) {
- PreparedStatement statement = connection.prepareStatement(sql);
+ try (Connection connection = connectionProvider.getConnection();
+  PreparedStatement statement = connection.prepareStatement(sql)) {
  int count = 0;
  for (RowData rowData : rowDataList) {
  setStatementData(statement, rowData, fieldGetters);
@@ -238,9 +238,9 @@ private void setStatementData(
  throws SQLException {
  if (row != null) {
  int index = 1;
- for (int i = 0; i < fieldGetters.length; i++) {
- for (int j = 0; j < fieldGetters[i].length; j++) {
- Object obj = fieldGetters[i][j].getFieldOrNull(row);
+ for (RowData.FieldGetter[] fieldGetter : fieldGetters) {
+ for (RowData.FieldGetter getter : fieldGetter) {
+ Object obj = getter.getFieldOrNull(row);
  if (obj == null) {
  statement.setObject(index++, null);
  } else {

flink-connector-oceanbase/src/main/java/com/oceanbase/connector/flink/sink/OceanBaseSink.java

@@ -22,6 +22,7 @@
 import com.oceanbase.connector.flink.dialect.OceanBaseMySQLDialect;
 import com.oceanbase.connector.flink.dialect.OceanBaseOracleDialect;
 import com.oceanbase.connector.flink.table.OceanBaseTableSchema;
+import org.apache.commons.lang3.StringUtils;
 
 import java.io.IOException;
 import java.sql.SQLException;
@@ -65,11 +66,11 @@ private OceanBaseDialect getDialect(OceanBaseConnectionProvider connectionProvid
  throw new IOException("Failed to get compatible mode", e);
  }
 
- if (compatibleMode != null) {
- compatibleMode = compatibleMode.toLowerCase();
+ if (StringUtils.isBlank(compatibleMode)) {
+ throw new RuntimeException("Got empty 'ob_compatibility_mode'");
  }
 
- switch (compatibleMode) {
+ switch (compatibleMode.toLowerCase()) {
  case "mysql":
  return new OceanBaseMySQLDialect();
  case "oracle":

flink-connector-oceanbase/src/main/java/com/oceanbase/connector/flink/sink/OceanBaseWriter.java

@@ -34,8 +34,9 @@ public class OceanBaseWriter<T> implements SinkWriter<T> {
  private final OceanBaseWriterOptions writerOptions;
  private final OceanBaseStatementExecutor<T> statementExecutor;
 
- private transient ScheduledExecutorService scheduler;
- private transient ScheduledFuture<?> scheduledFuture;
+ private final transient ScheduledExecutorService scheduler;
+ private final transient ScheduledFuture<?> scheduledFuture;
+
  private transient int bufferCount = 0;
  private transient volatile Exception flushException = null;
  private transient volatile boolean closed = false;
@@ -106,7 +107,7 @@ public synchronized void flush(boolean endOfInput) throws IOException, Interrupt
  if (i >= writerOptions.getMaxRetries()) {
  throw new IOException(e);
  }
- Thread.sleep(1000 * i);
+ Thread.sleep(1000L * i);
  }
  }
  }

flink-connector-oceanbase/src/main/java/com/oceanbase/connector/flink/table/OceanBaseConnectorOptions.java

@@ -122,9 +122,9 @@ public OceanBaseWriterOptions getWriterOptions() {
  allConfig.get(TABLE_NAME),
  allConfig.get(UPSERT_MODE),
  allConfig.get(BUFFER_FLUSH_INTERVAL).toMillis(),
- allConfig.get(BUFFER_SIZE).intValue(),
- allConfig.get(BUFFER_BATCH_SIZE).intValue(),
- allConfig.get(MAX_RETRIES).intValue());
+ allConfig.get(BUFFER_SIZE),
+ allConfig.get(BUFFER_BATCH_SIZE),
+ allConfig.get(MAX_RETRIES));
  }
 
  private Properties parseProperties(String propsStr) {