Repository: NetEase/arctic (java)

Issue:
[Subtask][Flink]: turn on all unit tests annotated with @Ignore 
### Search before asking

- [X] I have searched in the [issues](https://github.com/NetEase/arctic/issues?q=is%3Aissue) and found no similar issues.


### Description

improve unit test in Flink module with turning on all unit tests annotated with @Ignore.

### Parent issue

#1598 

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)


Pull request description:
[ARCTIC-1596][Flink]: turn on all unit tests annotated with @Ignore
## Why are the changes needed?

fix #1596

## Brief change log

  - *remove @Ignore in Flink module unit test*
  - *The flink 1.12, 1.14 and 1.15 versions are affected.*

## How was this patch tested?
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] Run test locally before making a pull request

## Documentation

  - Does this pull request introduces a new feature? (yes / **no**)
  - If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)


Patch:
diff --git a/ams/server/src/main/resources/mysql/ams-mysql-init.sql b/ams/server/src/main/resources/mysql/ams-mysql-init.sql
index 978592e40..94fc026ab 100644
--- a/ams/server/src/main/resources/mysql/ams-mysql-init.sql
+++ b/ams/server/src/main/resources/mysql/ams-mysql-init.sql
@@ -102,9 +102,9 @@ CREATE TABLE `table_runtime`
     `current_change_snapshotId`     bigint(20) DEFAULT NULL COMMENT 'Change table current snapshot id',
     `last_optimized_snapshotId`     bigint(20) NOT NULL DEFAULT '-1' COMMENT 'last optimized snapshot id',
     `last_optimized_change_snapshotId`     bigint(20) NOT NULL DEFAULT '-1' COMMENT 'last optimized change snapshot id',
-    `last_major_optimizing_time`    timestamp COMMENT 'Latest Major Optimize time for all partitions',
-    `last_minor_optimizing_time`    timestamp COMMENT 'Latest Minor Optimize time for all partitions',
-    `last_full_optimizing_time`     timestamp COMMENT 'Latest Full Optimize time for all partitions',
+    `last_major_optimizing_time`    timestamp NULL DEFAULT NULL COMMENT 'Latest Major Optimize time for all partitions',
+    `last_minor_optimizing_time`    timestamp NULL DEFAULT NULL COMMENT 'Latest Minor Optimize time for all partitions',
+    `last_full_optimizing_time`     timestamp NULL DEFAULT NULL COMMENT 'Latest Full Optimize time for all partitions',
     `optimizing_status`             varchar(20) DEFAULT 'Idle' COMMENT 'Table optimize status: MajorOptimizing, MinorOptimizing, Pending, Idle',
     `optimizing_status_start_time`  timestamp default CURRENT_TIMESTAMP COMMENT 'Table optimize status start time',
     `optimizing_process_id`         bigint(20) NOT NULL COMMENT 'optimizing_procedure UUID',
@@ -126,8 +126,8 @@ CREATE TABLE `table_optimizing_process`
     `target_change_snapshot_id`     bigint(20) NOT NULL,
     `status`                        varchar(10) NOT NULL COMMENT 'Direct to TableOptimizingStatus',
     `optimizing_type`               varchar(10) NOT NULL COMMENT 'Optimize type: Major, Minor',
-    `plan_time`                     timestamp default CURRENT_TIMESTAMP COMMENT 'First plan time',
-    `end_time`                      datetime(3) DEFAULT NULL COMMENT 'finish time or failed time',
+    `plan_time`                     timestamp DEFAULT CURRENT_TIMESTAMP COMMENT 'First plan time',
+    `end_time`                      timestamp NULL DEFAULT NULL COMMENT 'finish time or failed time',
     `fail_reason`                   varchar(4096) DEFAULT NULL COMMENT 'Error message after task failed',
     `rewrite_input`                 longblob DEFAULT NULL COMMENT 'rewrite files input',
     `summary`                       mediumtext COMMENT 'Max change transaction id of these tasks',
@@ -144,9 +144,9 @@ CREATE TABLE `task_runtime`
     `retry_num`                 int(11) DEFAULT NULL COMMENT 'Retry times',
     `table_id`                  bigint(20) NOT NULL,
     `partition_data`            varchar(128)  DEFAULT NULL COMMENT 'Partition data',
-    `create_time`               datetime(3) DEFAULT NULL COMMENT 'Task create time',
-    `start_time`                datetime(3) DEFAULT NULL COMMENT 'Time when task start waiting to execute',
-    `end_time`                  datetime(3) DEFAULT NULL COMMENT 'Time when task finished',
+    `create_time`               timestamp NULL DEFAULT NULL COMMENT 'Task create time',
+    `start_time`                timestamp NULL DEFAULT NULL COMMENT 'Time when task start waiting to execute',
+    `end_time`                  timestamp NULL DEFAULT NULL COMMENT 'Time when task finished',
     `cost_time`                 bigint(20) DEFAULT NULL,
     `status`                    varchar(16)   DEFAULT NULL  COMMENT 'Optimize Status: Init, Pending, Executing, Failed, Prepared, Committed',
     `fail_reason`               varchar(4096) DEFAULT NULL COMMENT 'Error message after task failed',
@@ -178,7 +178,7 @@ CREATE TABLE `api_tokens`
     `id`         int(11) NOT NULL AUTO_INCREMENT,
     `apikey`     varchar(256) NOT NULL COMMENT 'openapi client public key',
     `secret`     varchar(256) NOT NULL COMMENT 'The key used by the client to generate the request signature',
-    `apply_time` datetime DEFAULT NULL COMMENT 'apply time',
+    `apply_time` timestamp NULL DEFAULT NULL COMMENT 'apply time',
     PRIMARY KEY (`id`) USING BTREE,
     UNIQUE KEY `account_unique` (`apikey`) USING BTREE COMMENT 'account unique'
 ) ENGINE=InnoDB AUTO_INCREMENT=33 DEFAULT CHARSET=utf8mb4 ROW_FORMAT=DYNAMIC COMMENT='Openapi  secret';
@@ -198,8 +198,8 @@ CREATE TABLE `table_blocker` (
   `db_name` varchar(128) NOT NULL COMMENT 'Database name',
   `table_name` varchar(128) NOT NULL COMMENT 'Table name',
   `operations` varchar(128) NOT NULL COMMENT 'Blocked operations',
-  `create_time` datetime(3) DEFAULT NULL COMMENT 'Blocker create time',
-  `expiration_time` datetime(3) DEFAULT NULL COMMENT 'Blocker expiration time',
+  `create_time` timestamp NULL DEFAULT NULL COMMENT 'Blocker create time',
+  `expiration_time` timestamp NULL DEFAULT NULL COMMENT 'Blocker expiration time',
   `properties` mediumtext COMMENT 'Blocker properties',
   PRIMARY KEY (`blocker_id`),
   KEY `table_index` (`catalog_name`,`db_name`,`table_name`)
diff --git a/ams/server/src/main/resources/mysql/upgrade-0.4.1-to-0.5.0.sql b/ams/server/src/main/resources/mysql/upgrade-0.4.1-to-0.5.0.sql
index 337acb83f..1e3a67a88 100644
--- a/ams/server/src/main/resources/mysql/upgrade-0.4.1-to-0.5.0.sql
+++ b/ams/server/src/main/resources/mysql/upgrade-0.4.1-to-0.5.0.sql
@@ -94,9 +94,9 @@ CREATE TABLE `table_runtime`
     `current_change_snapshotId`     bigint(20) DEFAULT NULL COMMENT 'Change table current snapshot id',
     `last_optimized_snapshotId`     bigint(20) NOT NULL DEFAULT '-1' COMMENT 'last optimized snapshot id',
     `last_optimized_change_snapshotId`     bigint(20) NOT NULL DEFAULT '-1' COMMENT 'last optimized change snapshot id',
-    `last_major_optimizing_time`    timestamp COMMENT 'Latest Major Optimize time for all partitions',
-    `last_minor_optimizing_time`    timestamp COMMENT 'Latest Minor Optimize time for all partitions',
-    `last_full_optimizing_time`     timestamp COMMENT 'Latest Full Optimize time for all partitions',
+    `last_major_optimizing_time`    timestamp NULL DEFAULT NULL COMMENT 'Latest Major Optimize time for all partitions',
+    `last_minor_optimizing_time`    timestamp NULL DEFAULT NULL COMMENT 'Latest Minor Optimize time for all partitions',
+    `last_full_optimizing_time`     timestamp NULL DEFAULT NULL COMMENT 'Latest Full Optimize time for all partitions',
     `optimizing_status`             varchar(20) DEFAULT 'Idle' COMMENT 'Table optimize status: MajorOptimizing, MinorOptimizing, Pending, Idle',
     `optimizing_status_start_time`  timestamp default CURRENT_TIMESTAMP COMMENT 'Table optimize status start time',
     `optimizing_process_id`         bigint(20) NOT NULL COMMENT 'optimizing_procedure UUID',
@@ -116,9 +116,9 @@ CREATE TABLE `task_runtime`
     `retry_num`                 int(11) DEFAULT NULL COMMENT 'Retry times',
     `table_id`                  bigint(20) NOT NULL,
     `partition_data`            varchar(128)  DEFAULT NULL COMMENT 'Partition data',
-    `create_time`               datetime(3) DEFAULT NULL COMMENT 'Task create time',
-    `start_time`                datetime(3) DEFAULT NULL COMMENT 'Time when task start waiting to execute',
-    `end_time`                  datetime(3) DEFAULT NULL COMMENT 'Time when task finished',
+    `create_time`               timestamp NULL DEFAULT NULL COMMENT 'Task create time',
+    `start_time`                timestamp NULL DEFAULT NULL COMMENT 'Time when task start waiting to execute',
+    `end_time`                  timestamp NULL DEFAULT NULL COMMENT 'Time when task finished',
     `cost_time`                 bigint(20) DEFAULT NULL,
     `status`                    varchar(16)   DEFAULT NULL  COMMENT 'Optimize Status: Init, Pending, Executing, Failed, Prepared, Committed',
     `fail_reason`               varchar(4096) DEFAULT NULL COMMENT 'Error message after task failed',
@@ -143,8 +143,8 @@ CREATE TABLE `table_optimizing_process`
     `target_change_snapshot_id`     bigint(20) NOT NULL,
     `status`                        varchar(10) NOT NULL COMMENT 'Direct to TableOptimizingStatus',
     `optimizing_type`               varchar(10) NOT NULL COMMENT 'Optimize type: Major, Minor',
-    `plan_time`                     timestamp default CURRENT_TIMESTAMP COMMENT 'First plan time',
-    `end_time`                      datetime(3) DEFAULT NULL COMMENT 'finish time or failed time',
+    `plan_time`                     timestamp DEFAULT CURRENT_TIMESTAMP COMMENT 'First plan time',
+    `end_time`                      timestamp NULL DEFAULT NULL COMMENT 'finish time or failed time',
     `fail_reason`                   varchar(4096) DEFAULT NULL COMMENT 'Error message after task failed',
     `rewrite_input`                 longblob DEFAULT NULL COMMENT 'rewrite files input',
     `summary`                       mediumtext COMMENT 'Max change transaction id of these tasks',
@@ -168,6 +168,11 @@ CREATE TABLE `optimizing_task_quota`
     KEY  `table_index` (`table_id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT 'Optimize task basic information';
 
+-- modify to timestamp
+ALTER TABLE `table_blocker` MODIFY `create_time` timestamp NULL DEFAULT NULL COMMENT 'Blocker create time';
+ALTER TABLE `table_blocker` MODIFY `expiration_time` timestamp NULL DEFAULT NULL COMMENT 'Blocker expiration time';
+ALTER TABLE `api_tokens` MODIFY `apply_time` timestamp NULL DEFAULT NULL COMMENT 'apply time';
+
 -- init table_runtime
 INSERT INTO table_runtime (table_id, catalog_name, db_name, table_name, current_snapshot_id, current_change_snapshotId,
 last_optimized_snapshotId, last_optimized_change_snapshotId,
diff --git a/core/src/main/java/com/netease/arctic/utils/ArcticDataFiles.java b/core/src/main/java/com/netease/arctic/utils/ArcticDataFiles.java
index c2d42f443..826f03e79 100644
--- a/core/src/main/java/com/netease/arctic/utils/ArcticDataFiles.java
+++ b/core/src/main/java/com/netease/arctic/utils/ArcticDataFiles.java
@@ -114,6 +114,10 @@ public class ArcticDataFiles {
               field.name().equals(parts[0]),
           "Invalid partition: %s", partitions[i]);
 
+      if ("null".equals(parts[1])) {
+        parts[1] = null;
+      }
+
       data.set(i, ArcticDataFiles.fromPartitionString(field, spec.partitionType().fieldType(parts[0]), parts[1]));
     }
 


Test patch:
diff --git a/ams/server/src/test/java/com/netease/arctic/server/optimizing/flow/view/KeyedTableDataView.java b/ams/server/src/test/java/com/netease/arctic/server/optimizing/flow/view/KeyedTableDataView.java
index 70159bd29..c9c21c997 100644
--- a/ams/server/src/test/java/com/netease/arctic/server/optimizing/flow/view/KeyedTableDataView.java
+++ b/ams/server/src/test/java/com/netease/arctic/server/optimizing/flow/view/KeyedTableDataView.java
@@ -248,7 +248,11 @@ public class KeyedTableDataView extends AbstractTableDataView {
       Object o1 = r1.get(i);
       Object o2 = r2.get(i);
       boolean equals;
-      if (o1 instanceof OffsetDateTime) {
+      if (o1 == null && o2 == null) {
+        return true;
+      } else if (o1 == null || o2 == null) {
+        return false;
+      } else if (o1 instanceof OffsetDateTime) {
         equals = ((OffsetDateTime) o1).isEqual((OffsetDateTime) o2);
       } else {
         equals = o1.equals(o2);
diff --git a/core/src/test/java/com/netease/arctic/utils/TestArcticDataFiles.java b/core/src/test/java/com/netease/arctic/utils/TestArcticDataFiles.java
index c9a592770..4f83a47f0 100644
--- a/core/src/test/java/com/netease/arctic/utils/TestArcticDataFiles.java
+++ b/core/src/test/java/com/netease/arctic/utils/TestArcticDataFiles.java
@@ -128,4 +128,16 @@ public class TestArcticDataFiles {
     p2.set(partitionData);
     Assert.assertEquals(p1, p2);
   }
+
+  @Test
+  public void testNullPartition() {
+    Schema schema = new Schema(
+        Types.NestedField.required(1, "name", Types.StringType.get())
+    );
+    PartitionSpec spec = PartitionSpec.builderFor(schema).identity("name").build();
+    PartitionKey partitionKey = new PartitionKey(spec, schema);
+    String partitionPath = "name=null";
+    StructLike partitionData = ArcticDataFiles.data(spec, partitionPath);
+    Assert.assertNull(partitionData.get(0, Types.StringType.get().typeId().javaClass()));
+  }
 }
diff --git a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/read/TestArcticSource.java b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/read/TestArcticSource.java
index 5f8231450..799b9e0aa 100644
--- a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/read/TestArcticSource.java
+++ b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/read/TestArcticSource.java
@@ -77,7 +77,6 @@ import org.apache.iceberg.types.Types;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -175,13 +174,11 @@ public class TestArcticSource extends TestRowDataReaderFunction implements Seria
     assertArrayEquals(excepts(), actualResult);
   }
 
-  @Ignore
   @Test
   public void testArcticSourceStaticJobManagerFailover() throws Exception {
     testArcticSource(FailoverTestUtil.FailoverType.JM);
   }
 
-  @Ignore
   @Test
   public void testArcticSourceStaticTaskManagerFailover() throws Exception {
     testArcticSource(FailoverTestUtil.FailoverType.TM);
@@ -516,13 +513,11 @@ public class TestArcticSource extends TestRowDataReaderFunction implements Seria
     jobClient.cancel();
   }
 
-  @Ignore
   @Test
   public void testArcticContinuousSourceJobManagerFailover() throws Exception {
     testArcticContinuousSource(FailoverTestUtil.FailoverType.JM);
   }
 
-  @Ignore
   @Test
   public void testArcticContinuousSourceTaskManagerFailover() throws Exception {
     testArcticContinuousSource(FailoverTestUtil.FailoverType.TM);
diff --git a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/read/hidden/kafka/TestKafkaConsumerBaseLog.java b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/read/hidden/kafka/TestKafkaConsumerBaseLog.java
index df4078644..edf3356b2 100644
--- a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/read/hidden/kafka/TestKafkaConsumerBaseLog.java
+++ b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/read/hidden/kafka/TestKafkaConsumerBaseLog.java
@@ -18,6 +18,7 @@
 
 package com.netease.arctic.flink.read.hidden.kafka;
 
+import com.netease.arctic.flink.util.TestUtil;
 import com.netease.arctic.flink.util.kafka.KafkaTestBase;
 import com.netease.arctic.flink.write.hidden.TestBaseLog;
 import org.apache.flink.streaming.connectors.kafka.internals.FlinkKafkaInternalProducer;
@@ -27,10 +28,12 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
 import org.junit.AfterClass;
+import org.junit.Before;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
+import org.junit.Rule;
 import org.junit.Test;
 import org.junit.jupiter.api.Assertions;
+import org.junit.rules.TestName;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -45,14 +48,15 @@ import java.util.stream.Collectors;
 
 import static com.netease.arctic.flink.util.kafka.KafkaConfigGenerate.getProperties;
 import static com.netease.arctic.flink.util.kafka.KafkaConfigGenerate.getPropertiesWithByteArray;
-import static com.netease.arctic.flink.write.hidden.TestHiddenLogOperators.topic;
 import static org.apache.kafka.clients.producer.ProducerConfig.TRANSACTIONAL_ID_CONFIG;
 import static org.assertj.core.api.Assertions.assertThat;
 
-@Ignore
 public class TestKafkaConsumerBaseLog extends TestBaseLog {
   private static final Logger LOG = LoggerFactory.getLogger(TestKafkaConsumerBaseLog.class);
   private static final KafkaTestBase kafkaTestBase = new KafkaTestBase();
+  @Rule
+  public TestName testName = new TestName();
+  private static String topic;
 
   @BeforeClass
   public static void prepare() throws Exception {
@@ -64,6 +68,11 @@ public class TestKafkaConsumerBaseLog extends TestBaseLog {
     kafkaTestBase.shutDownServices();
   }
 
+  @Before
+  public void before() {
+    topic = TestUtil.getUtMethodName(testName);
+  }
+
   @Test
   public void testTransactionalConsume() {
     final String topic = "test-offset-flip";
diff --git a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/TestAutomaticLogWriter.java b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/TestAutomaticLogWriter.java
index 9438beade..2d815842f 100644
--- a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/TestAutomaticLogWriter.java
+++ b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/TestAutomaticLogWriter.java
@@ -58,7 +58,6 @@ import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.jupiter.api.Assertions;
 import org.junit.runner.RunWith;
@@ -87,7 +86,6 @@ import static com.netease.arctic.table.TableProperties.ENABLE_LOG_STORE;
 import static com.netease.arctic.table.TableProperties.LOG_STORE_ADDRESS;
 import static com.netease.arctic.table.TableProperties.LOG_STORE_MESSAGE_TOPIC;
 
-@Ignore
 @RunWith(Parameterized.class)
 public class TestAutomaticLogWriter extends FlinkTestBase {
   private static final Logger LOG = LoggerFactory.getLogger(TestAutomaticLogWriter.class);
diff --git a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/hidden/kafka/TestHiddenKafkaProducer.java b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/hidden/kafka/TestHiddenKafkaProducer.java
index e292652c4..0257262b0 100644
--- a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/hidden/kafka/TestHiddenKafkaProducer.java
+++ b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/hidden/kafka/TestHiddenKafkaProducer.java
@@ -38,7 +38,6 @@ import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -69,7 +68,6 @@ public class TestHiddenKafkaProducer extends TestBaseLog {
     kafkaTestBase.shutDownServices();
   }
 
-  @Ignore
   @Test
   public void testInitTransactionId() {
     final String topic = "test-init-transactions";
diff --git a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/read/TestArcticSource.java b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/read/TestArcticSource.java
index af8f291a7..b7291d144 100644
--- a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/read/TestArcticSource.java
+++ b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/read/TestArcticSource.java
@@ -77,7 +77,6 @@ import org.apache.iceberg.types.Types;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -174,13 +173,11 @@ public class TestArcticSource extends TestRowDataReaderFunction implements Seria
     assertArrayEquals(excepts(), actualResult);
   }
 
-  @Ignore
   @Test
   public void testArcticSourceStaticJobManagerFailover() throws Exception {
     testArcticSource(FailoverType.JM);
   }
 
-  @Ignore
   @Test
   public void testArcticSourceStaticTaskManagerFailover() throws Exception {
     testArcticSource(FailoverType.TM);
@@ -517,13 +514,11 @@ public class TestArcticSource extends TestRowDataReaderFunction implements Seria
     jobClient.cancel();
   }
 
-  @Ignore
   @Test
   public void testArcticContinuousSourceJobManagerFailover() throws Exception {
     testArcticContinuousSource(FailoverType.JM);
   }
 
-  @Ignore
   @Test
   public void testArcticContinuousSourceTaskManagerFailover() throws Exception {
     testArcticContinuousSource(FailoverType.TM);
diff --git a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/read/hidden/kafka/TestKafkaConsumer.java b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/read/hidden/kafka/TestKafkaConsumer.java
index 1907eae7b..cdde05c5e 100644
--- a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/read/hidden/kafka/TestKafkaConsumer.java
+++ b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/read/hidden/kafka/TestKafkaConsumer.java
@@ -28,7 +28,6 @@ import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.jupiter.api.Assertions;
 import org.slf4j.Logger;
@@ -49,7 +48,6 @@ import static com.netease.arctic.flink.write.hidden.kafka.TestHiddenLogOperators
 import static org.apache.kafka.clients.producer.ProducerConfig.TRANSACTIONAL_ID_CONFIG;
 import static org.assertj.core.api.Assertions.assertThat;
 
-@Ignore
 public class TestKafkaConsumer extends TestBaseLog {
   private static final Logger LOG = LoggerFactory.getLogger(TestKafkaConsumer.class);
   private static final KafkaTestBase kafkaTestBase = new KafkaTestBase();
diff --git a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/TestAutomaticLogWriter.java b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/TestAutomaticLogWriter.java
index ea9064163..2e64a5860 100644
--- a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/TestAutomaticLogWriter.java
+++ b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/TestAutomaticLogWriter.java
@@ -58,7 +58,6 @@ import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.jupiter.api.Assertions;
 import org.junit.runner.RunWith;
@@ -87,8 +86,6 @@ import static com.netease.arctic.table.TableProperties.ENABLE_LOG_STORE;
 import static com.netease.arctic.table.TableProperties.LOG_STORE_ADDRESS;
 import static com.netease.arctic.table.TableProperties.LOG_STORE_MESSAGE_TOPIC;
 
-//Enable when change to new Test Base
-@Ignore
 @RunWith(Parameterized.class)
 public class TestAutomaticLogWriter extends FlinkTestBase {
   private static final Logger LOG = LoggerFactory.getLogger(TestAutomaticLogWriter.class);
diff --git a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/hidden/kafka/TestHiddenKafkaProducer.java b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/hidden/kafka/TestHiddenKafkaProducer.java
index 303f5448f..393c47da5 100644
--- a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/hidden/kafka/TestHiddenKafkaProducer.java
+++ b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/hidden/kafka/TestHiddenKafkaProducer.java
@@ -37,7 +37,6 @@ import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -68,7 +67,6 @@ public class TestHiddenKafkaProducer extends TestBaseLog {
     kafkaTestBase.shutDownServices();
   }
 
-  @Ignore
   @Test
   public void testInitTransactionId() {
     final String topic = "test-init-transactions";
diff --git a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/read/TestArcticSource.java b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/read/TestArcticSource.java
index e39db731d..4d789a1ef 100644
--- a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/read/TestArcticSource.java
+++ b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/read/TestArcticSource.java
@@ -77,7 +77,6 @@ import org.apache.iceberg.types.Types;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -175,13 +174,11 @@ public class TestArcticSource extends TestRowDataReaderFunction implements Seria
     assertArrayEquals(excepts(), actualResult);
   }
 
-  @Ignore
   @Test
   public void testArcticSourceStaticJobManagerFailover() throws Exception {
     testArcticSource(FailoverType.JM);
   }
 
-  @Ignore
   @Test
   public void testArcticSourceStaticTaskManagerFailover() throws Exception {
     testArcticSource(FailoverType.TM);
@@ -517,13 +514,11 @@ public class TestArcticSource extends TestRowDataReaderFunction implements Seria
     jobClient.cancel();
   }
 
-  @Ignore
   @Test
   public void testArcticContinuousSourceJobManagerFailover() throws Exception {
     testArcticContinuousSource(FailoverType.JM);
   }
 
-  @Ignore
   @Test
   public void testArcticContinuousSourceTaskManagerFailover() throws Exception {
     testArcticContinuousSource(FailoverType.TM);
diff --git a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/read/hidden/kafka/TestKafkaConsumer.java b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/read/hidden/kafka/TestKafkaConsumer.java
index 1907eae7b..cdde05c5e 100644
--- a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/read/hidden/kafka/TestKafkaConsumer.java
+++ b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/read/hidden/kafka/TestKafkaConsumer.java
@@ -28,7 +28,6 @@ import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.jupiter.api.Assertions;
 import org.slf4j.Logger;
@@ -49,7 +48,6 @@ import static com.netease.arctic.flink.write.hidden.kafka.TestHiddenLogOperators
 import static org.apache.kafka.clients.producer.ProducerConfig.TRANSACTIONAL_ID_CONFIG;
 import static org.assertj.core.api.Assertions.assertThat;
 
-@Ignore
 public class TestKafkaConsumer extends TestBaseLog {
   private static final Logger LOG = LoggerFactory.getLogger(TestKafkaConsumer.class);
   private static final KafkaTestBase kafkaTestBase = new KafkaTestBase();
diff --git a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/TestAutomaticLogWriter.java b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/TestAutomaticLogWriter.java
index 442d5f407..ac435db5a 100644
--- a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/TestAutomaticLogWriter.java
+++ b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/TestAutomaticLogWriter.java
@@ -58,7 +58,6 @@ import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.jupiter.api.Assertions;
 import org.junit.runner.RunWith;
@@ -88,8 +87,6 @@ import static com.netease.arctic.table.TableProperties.ENABLE_LOG_STORE;
 import static com.netease.arctic.table.TableProperties.LOG_STORE_ADDRESS;
 import static com.netease.arctic.table.TableProperties.LOG_STORE_MESSAGE_TOPIC;
 
-//Enable when change to new Test Base
-@Ignore
 @RunWith(Parameterized.class)
 public class TestAutomaticLogWriter extends FlinkTestBase {
   private static final Logger LOG = LoggerFactory.getLogger(TestAutomaticLogWriter.class);
diff --git a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/hidden/kafka/TestHiddenKafkaProducer.java b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/hidden/kafka/TestHiddenKafkaProducer.java
index 303f5448f..393c47da5 100644
--- a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/hidden/kafka/TestHiddenKafkaProducer.java
+++ b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/hidden/kafka/TestHiddenKafkaProducer.java
@@ -37,7 +37,6 @@ import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -68,7 +67,6 @@ public class TestHiddenKafkaProducer extends TestBaseLog {
     kafkaTestBase.shutDownServices();
   }
 
-  @Ignore
   @Test
   public void testInitTransactionId() {
     final String topic = "test-init-transactions";