[Github Action] Fix flaky test (apache#5982)

*Motivation*

Some tests are failed by the file they used was changes.

.github/workflows/ci-unit-flaky.yaml

@@ -46,9 +46,21 @@ jobs:
         with:
           maven-version: 3.6.1
 
+      - name: run PersistentTransactionBuffer test
+        run: mvn -B -am -pl pulsar-broker -Dorg.slf4j.simpleLogger.defaultLogLevel=ERROR clean install -Dtest=PersistentTransactionBufferTest -DfailIfNoTests=false
+
+      - name: run ServerCnx test
+        run: mvn -B -am -pl pulsar-broker -Dorg.slf4j.simpleLogger.defaultLogLevel=ERROR clean install -Dtest=ServerCnxTest -DfailIfNoTests=false
+
+      - name: run MessagePublishThrottling test
+        run: mvn -B -am -pl pulsar-broker -Dorg.slf4j.simpleLogger.defaultLogLevel=ERROR clean install -Dtest=MessagePublishThrottlingTest -DfailIfNoTests=false
+
+      - name: run PrimitiveSchema test
+        run: mvn -B -am -pl pulsar-client -Dorg.slf4j.simpleLogger.defaultLogLevel=ERROR clean install -Dtest=PrimitiveSchemaTest -DfailIfNoTests=false
+
       - name: run unit tests
-        run: mvn -B -Dorg.slf4j.simpleLogger.defaultLogLevel=ERROR install '-Dtest=PersistentTransactionBufferTest,PulsarFunctionE2ESecurityTest,ServerCnxTest,AdminApiOffloadTest,AdminApiSchemaValidationEnforced,V1_AdminApiTest2,ProxyPublishConsumeTlsTest,PulsarFunctionE2ETest,MessageIdSerialization,AdminApiTest2,PulsarFunctionLocalRunTest,PartitionedProducerConsumerTest,KafkaProducerSimpleConsumerTest,ProxyTest' -DfailIfNoTests=false
-          
+        run: mvn -B -Dorg.slf4j.simpleLogger.defaultLogLevel=ERROR clean install '-Dtest=PulsarFunctionE2ESecurityTest,AdminApiOffloadTest,AdminApiSchemaValidationEnforced,V1_AdminApiTest2,PulsarFunctionE2ETest,MessageIdSerialization,AdminApiTest2,PulsarFunctionLocalRunTest,PartitionedProducerConsumerTest,KafkaProducerSimpleConsumerTest,ProxyTest' -DfailIfNoTests=false
+
       - name: package surefire artifacts
         if: failure()
         run: |

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaValidationEnforced.java

@@ -148,7 +148,7 @@ public void testEnableSchemaValidationEnforcedHasSchemaMismatch() throws Excepti
         schemaInfo.setProperties(properties);
         schemaInfo.setName("test");
         schemaInfo.setSchema("".getBytes());
-        PostSchemaPayload postSchemaPayload = new PostSchemaPayload("STRING", "{'key':'value'}", properties);
+        PostSchemaPayload postSchemaPayload = new PostSchemaPayload("STRING", "", properties);
         admin.schemas().createSchema(topicName, postSchemaPayload);
         try (Producer p = pulsarClient.newProducer().topic(topicName).create()) {
             fail("Client no schema, but topic has schema, should fail");

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java

@@ -301,7 +301,7 @@ public void testSetPersistencepolicies() throws Exception {
         final String namespace = "prop-xyz/ns2";
         admin.namespaces().createNamespace(namespace, Sets.newHashSet("test"));
 
-        assertEquals(admin.namespaces().getPersistence(namespace), new PersistencePolicies(1, 1, 1, 0.0));
+        assertEquals(admin.namespaces().getPersistence(namespace), new PersistencePolicies(2, 2, 2, 0.0));
         admin.namespaces().setPersistence(namespace, new PersistencePolicies(3, 3, 3, 10.0));
         assertEquals(admin.namespaces().getPersistence(namespace), new PersistencePolicies(3, 3, 3, 10.0));
 
@@ -866,8 +866,8 @@ public void brokerNamespaceIsolationPolicies() throws Exception {
 
         // create
         String policyName1 = "policy-1";
-        String namespaceRegex = "other/use/other.*";
-        String cluster = "use";
+        String cluster = pulsar.getConfiguration().getClusterName();
+        String namespaceRegex = "other/" + cluster + "/other.*";
         String brokerName = pulsar.getAdvertisedAddress();
         String brokerAddress = brokerName + ":" + pulsar.getConfiguration().getWebServicePort().get();
         NamespaceIsolationData nsPolicyData1 = new NamespaceIsolationData();
@@ -897,11 +897,8 @@ public void brokerNamespaceIsolationPolicies() throws Exception {
         assertEquals(brokerIsolationData.namespaceRegex.size(), 1);
         assertEquals(brokerIsolationData.namespaceRegex.get(0), namespaceRegex);
 
-        try {
-            admin.clusters().getBrokerWithNamespaceIsolationPolicy(cluster, "invalid-broker");
-            Assert.fail("should have failed due to invalid broker address");
-        } catch (PulsarAdminException.NotFoundException e) {// expected
-        }
+        BrokerNamespaceIsolationData isolationData = admin.clusters().getBrokerWithNamespaceIsolationPolicy(cluster, "invalid-broker");
+        assertFalse(isolationData.isPrimary);
     }
 
     @Test
@@ -924,8 +921,11 @@ public void testClusterIsReadyBeforeCreateTopic() throws PulsarAdminException {
         final String persistentPartitionedTopicName = "persistent://prop-xyz/ns2/" + topicName;
         final String NonPersistentPartitionedTopicName = "non-persistent://prop-xyz/ns2/" + topicName;
 
-        // init tenant and namespace without cluster
         admin.namespaces().createNamespace("prop-xyz/ns2");
+        // By default the cluster will configure as configuration file. So the create topic operation
+        // will never throw exception except there is no cluster.
+        admin.namespaces().setNamespaceReplicationClusters("prop-xyz/ns2", new HashSet<String>());
+
         try {
             admin.topics().createPartitionedTopic(persistentPartitionedTopicName, partitions);
             Assert.fail("should have failed due to Namespace does not have any clusters configured");

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest2.java

@@ -293,7 +293,7 @@ public void testSetPersistencepolicies() throws Exception {
         final String namespace = "prop-xyz/use/ns2";
         admin.namespaces().createNamespace(namespace);
 
-        assertEquals(admin.namespaces().getPersistence(namespace), new PersistencePolicies(1, 1, 1, 0.0));
+        assertEquals(admin.namespaces().getPersistence(namespace), new PersistencePolicies(2, 2, 2, 0.0));
         admin.namespaces().setPersistence(namespace, new PersistencePolicies(3, 3, 3, 10.0));
         assertEquals(admin.namespaces().getPersistence(namespace), new PersistencePolicies(3, 3, 3, 10.0));
 
@@ -707,13 +707,13 @@ public void namespaceAntiAffinity() throws PulsarAdminException {
         admin.namespaces().setNamespaceAntiAffinityGroup(ns3, antiAffinityGroup);
 
         Set<String> namespaces = new HashSet<>(
-                admin.namespaces().getAntiAffinityNamespaces("dummy", "use", antiAffinityGroup));
+                admin.namespaces().getAntiAffinityNamespaces("prop-xyz", "use", antiAffinityGroup));
         assertEquals(namespaces.size(), 3);
         assertTrue(namespaces.contains(ns1));
         assertTrue(namespaces.contains(ns2));
         assertTrue(namespaces.contains(ns3));
 
-        List<String> namespaces2 = admin.namespaces().getAntiAffinityNamespaces("dummy", "use", "invalid-group");
+        List<String> namespaces2 = admin.namespaces().getAntiAffinityNamespaces("prop-xyz", "use", "invalid-group");
         assertEquals(namespaces2.size(), 0);
     }
 

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageIdSerialization.java

@@ -48,7 +48,7 @@ void testProtobufSerializationNull() throws Exception {
         MessageId.fromByteArray(null);
     }
 
-    @Test(expectedExceptions = IOException.class)
+    @Test(expectedExceptions = RuntimeException.class)
     void testProtobufSerializationEmpty() throws Exception {
         MessageId.fromByteArray(new byte[0]);
     }

pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/ProxyServer.java

@@ -122,23 +122,23 @@ public void addRestResources(String basePath, String javaPackages, String attrib
 
     public void start() throws PulsarServerException {
         log.info("Starting web socket proxy at port {}", conf.getWebServicePort().get());
-        try {
-            RequestLogHandler requestLogHandler = new RequestLogHandler();
-            Slf4jRequestLog requestLog = new Slf4jRequestLog();
-            requestLog.setExtended(true);
-            requestLog.setLogTimeZone(TimeZone.getDefault().getID());
-            requestLog.setLogLatency(true);
-            requestLogHandler.setRequestLog(requestLog);
-            handlers.add(0, new ContextHandlerCollection());
-            handlers.add(requestLogHandler);
-
-            ContextHandlerCollection contexts = new ContextHandlerCollection();
-            contexts.setHandlers(handlers.toArray(new Handler[handlers.size()]));
-
-            HandlerCollection handlerCollection = new HandlerCollection();
-            handlerCollection.setHandlers(new Handler[] { contexts, new DefaultHandler(), requestLogHandler });
-            server.setHandler(handlerCollection);
+        RequestLogHandler requestLogHandler = new RequestLogHandler();
+        Slf4jRequestLog requestLog = new Slf4jRequestLog();
+        requestLog.setExtended(true);
+        requestLog.setLogTimeZone(TimeZone.getDefault().getID());
+        requestLog.setLogLatency(true);
+        requestLogHandler.setRequestLog(requestLog);
+        handlers.add(0, new ContextHandlerCollection());
+        handlers.add(requestLogHandler);
+
+        ContextHandlerCollection contexts = new ContextHandlerCollection();
+        contexts.setHandlers(handlers.toArray(new Handler[handlers.size()]));
+
+        HandlerCollection handlerCollection = new HandlerCollection();
+        handlerCollection.setHandlers(new Handler[] { contexts, new DefaultHandler(), requestLogHandler });
+        server.setHandler(handlerCollection);
 
+        try {
             server.start();
         } catch (Exception e) {
             throw new PulsarServerException(e);

pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/WebSocketProxyConfiguration.java

@@ -93,7 +93,7 @@ public class WebSocketProxyConfiguration implements PulsarConfiguration {
     private int webSocketNumIoThreads = Runtime.getRuntime().availableProcessors();
 
     // Number of threads to use in HTTP server
-    private int numHttpServerThreads = Math.max(4, Runtime.getRuntime().availableProcessors());
+    private int numHttpServerThreads = Math.max(6, Runtime.getRuntime().availableProcessors());
 
     // Number of connections per Broker in Pulsar Client used in WebSocket proxy
     private int webSocketConnectionsPerBroker = Runtime.getRuntime().availableProcessors();