[ISSUE #10373] Fix quarantined flaky tests and remove CI rerun workflow (#10379)

Author: lizhimins

.github/workflows/rerun-workflow.yml

@@ -61,7 +61,6 @@
 import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
-import org.junit.Ignore;
 
 @RunWith(MockitoJUnitRunner.class)
 public class TransactionalMessageServiceImplTest {
@@ -163,15 +162,14 @@ public void testDeletePrepareMessage_queueFull() throws InterruptedException {
         assertThat(res).isFalse();
     }
 
-    @Ignore("Flaky: fails 1/100 runs (1.0%)")
     @Test
     public void testDeletePrepareMessage_maxSize() throws InterruptedException {
         brokerController.getBrokerConfig().setTransactionOpMsgMaxSize(1);
         brokerController.getBrokerConfig().setTransactionOpBatchInterval(3000);
         queueTransactionMsgService.open();
         boolean res = queueTransactionMsgService.deletePrepareMessage(createMessageBrokerInner(1000, "test", "testHello"));
         assertThat(res).isTrue();
-        verify(bridge, timeout(50)).writeOp(any(Integer.class), any(Message.class));
+        verify(bridge, timeout(3000)).writeOp(any(Integer.class), any(Message.class));
         queueTransactionMsgService.close();
     }
 

broker/src/test/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImplTest.java

@@ -49,7 +49,6 @@
 import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
 import org.apache.rocketmq.client.impl.consumer.ProcessQueue;
 import org.apache.rocketmq.client.impl.consumer.PullAPIWrapper;
-import org.apache.rocketmq.client.impl.consumer.PullMessageService;
 import org.apache.rocketmq.client.impl.consumer.PullRequest;
 import org.apache.rocketmq.client.impl.consumer.PullResultExt;
 import org.apache.rocketmq.client.impl.consumer.RebalancePushImpl;
@@ -88,7 +87,6 @@
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.when;
-import org.junit.Ignore;
 
 @RunWith(MockitoJUnitRunner.Silent.class)
 public class DefaultMQConsumerWithTraceTest {
@@ -217,7 +215,6 @@ public void terminate() {
         pushConsumer.shutdown();
     }
 
-    @Ignore("Flaky: fails 1/100 runs (1.0%)")
     @Test
     public void testPullMessage_WithTrace_Success() throws InterruptedException, RemotingException, MQBrokerException, MQClientException {
         traceProducer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl());
@@ -234,9 +231,8 @@ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
             }
         }));
 
-        PullMessageService pullMessageService = mQClientFactory.getPullMessageService();
-        pullMessageService.executePullRequestImmediately(createPullRequest());
-        countDownLatch.await(30, TimeUnit.SECONDS);
+        pushConsumer.getDefaultMQPushConsumerImpl().pullMessage(createPullRequest());
+        countDownLatch.await(10, TimeUnit.SECONDS);
         MessageExt msg = messageAtomic.get();
         assertThat(msg).isNotNull();
         assertThat(msg.getTopic()).isEqualTo(topic);

client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQConsumerWithTraceTest.java

@@ -80,8 +80,6 @@
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.when;
-import org.junit.Ignore;
-
 @RunWith(MockitoJUnitRunner.class)
 public class DefaultMQLitePullConsumerWithTraceTest {
 
@@ -118,6 +116,11 @@ public void init() throws Exception {
                 MQClientManager.getInstance(), "factoryTable", true);
         factoryTable.forEach((clientId, instance) -> instance.shutdown());
         factoryTable.clear();
+
+        Field field = RebalanceService.class.getDeclaredField("waitInterval");
+        field.setAccessible(true);
+        field.set(null, 100L);
+
         mQClientFactory = null;
         mqClientInstance = null;
         traceMqClientInstance = null;
@@ -154,7 +157,6 @@ public void destroy() {
         }
     }
 
-    @Ignore("Flaky: fails 6/100 runs (6.0%)")
     @Test
     public void testSubscribe_PollMessageSuccess_WithDefaultTraceTopic() throws Exception {
         DefaultLitePullConsumer litePullConsumer = createLitePullConsumerWithDefaultTraceTopic();
@@ -171,7 +173,6 @@ public void testSubscribe_PollMessageSuccess_WithDefaultTraceTopic() throws Exce
         }
     }
 
-    @Ignore("Flaky: fails 5/100 runs (5.0%)")
     @Test
     public void testSubscribe_PollMessageSuccess_WithCustomizedTraceTopic() throws Exception {
         DefaultLitePullConsumer litePullConsumer = createLitePullConsumerWithCustomizedTraceTopic();
@@ -238,13 +239,6 @@ private void initDefaultLitePullConsumer(DefaultLitePullConsumer litePullConsume
         mQClientFactory.getClientConfig().setDecodeReadBody(true);
         field.set(litePullConsumerImpl, mQClientFactory);
 
-        field = MQClientInstance.class.getDeclaredField("rebalanceService");
-        field.setAccessible(true);
-        RebalanceService rebalanceService = (RebalanceService) field.get(mQClientFactory);
-        field = RebalanceService.class.getDeclaredField("waitInterval");
-        field.setAccessible(true);
-        field.set(rebalanceService, 100);
-
         PullAPIWrapper pullAPIWrapper = litePullConsumerImpl.getPullAPIWrapper();
         field = PullAPIWrapper.class.getDeclaredField("mQClientFactory");
         field.setAccessible(true);

client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQLitePullConsumerWithTraceTest.java

@@ -25,7 +25,6 @@
 import java.util.Random;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import org.junit.Ignore;
 
 public class BloomFilterTest {
 
@@ -143,14 +142,14 @@ public void testBloomFilterData() {
         assertThat(bloomFilter.isValid(bloomFilterData)).isFalse();
     }
 
-    @Ignore("Flaky: fails 1/100 runs (1.0%)")
     @Test
     public void testCheckFalseHit() {
+        Random random = new Random(42);
         BloomFilter bloomFilter = BloomFilter.createByFn(1, 300);
         BitsArray bits = BitsArray.create(bloomFilter.getM());
         int falseHit = 0;
         for (int i = 0; i < bloomFilter.getN(); i++) {
-            String str = randomString((new Random(System.nanoTime())).nextInt(127) + 10);
+            String str = randomString(random, random.nextInt(127) + 10);
             int[] bitPos = bloomFilter.calcBitPositions(str);
 
             if (bloomFilter.checkFalseHit(bitPos, bits)) {
@@ -163,10 +162,10 @@ public void testCheckFalseHit() {
         assertThat(falseHit).isLessThanOrEqualTo(bloomFilter.getF() * bloomFilter.getN() / 100);
     }
 
-    private String randomString(int length) {
+    private String randomString(Random random, int length) {
         StringBuilder stringBuilder = new StringBuilder(length);
         for (int i = 0; i < length; i++) {
-            stringBuilder.append((char) ((new Random(System.nanoTime())).nextInt(123 - 97) + 97));
+            stringBuilder.append((char) (random.nextInt(26) + 'a'));
         }
 
         return stringBuilder.toString();