[improve][test] remove powermock-reflect dependency (apache#17696)

pom.xml

@@ -241,7 +241,6 @@ flexible messaging model and an intuitive client API.</description>
     <kerby.version>1.1.1</kerby.version>
     <testng.version>7.3.0</testng.version>
     <mockito.version>3.12.4</mockito.version>
-    <powermock.version>2.0.9</powermock.version>
     <javassist.version>3.25.0-GA</javassist.version>
     <skyscreamer.version>1.5.0</skyscreamer.version>
     <objenesis.version>3.1</objenesis.version>
@@ -340,12 +339,6 @@ flexible messaging model and an intuitive client API.</description>
         <version>${mockito.version}</version>
       </dependency>
 
-      <dependency>
-        <groupId>org.powermock</groupId>
-        <artifactId>powermock-reflect</artifactId>
-        <version>${powermock.version}</version>
-      </dependency>
-
       <dependency>
         <groupId>org.apache.zookeeper</groupId>
         <artifactId>zookeeper</artifactId>
@@ -1331,12 +1324,6 @@ flexible messaging model and an intuitive client API.</description>
       <scope>test</scope>
     </dependency>
 
-    <dependency>
-      <groupId>org.powermock</groupId>
-      <artifactId>powermock-reflect</artifactId>
-      <scope>test</scope>
-    </dependency>
-
     <dependency>
       <groupId>org.assertj</groupId>
       <artifactId>assertj-core</artifactId>

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java

@@ -106,7 +106,7 @@ public CompletableFuture<OwnedBundle> asyncLoad(NamespaceBundle namespaceBundle,
                                 .thenRun(() -> {
                                     log.info("Resource lock for {} has expired", rl.getPath());
                                     namespaceService.unloadNamespaceBundle(namespaceBundle);
-                                    ownedBundlesCache.synchronous().invalidate(namespaceBundle);
+                                    invalidateLocalOwnerCache(namespaceBundle);
                                     namespaceService.onNamespaceBundleUnload(namespaceBundle);
                                 });
                         return new OwnedBundle(namespaceBundle);
@@ -330,6 +330,10 @@ public void invalidateLocalOwnerCache() {
         this.ownedBundlesCache.synchronous().invalidateAll();
     }
 
+    public void invalidateLocalOwnerCache(NamespaceBundle namespaceBundle) {
+        this.ownedBundlesCache.synchronous().invalidate(namespaceBundle);
+    }
+
     public synchronized boolean refreshSelfOwnerInfo() {
         this.selfOwnerInfo = new NamespaceEphemeralData(pulsar.getBrokerServiceUrl(),
                 pulsar.getBrokerServiceUrlTls(), pulsar.getSafeWebServiceAddress(),

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

@@ -1281,6 +1281,11 @@ public ManagedCursor getCursor() {
         return cursor;
     }
 
+    @VisibleForTesting
+    public PendingAckHandle getPendingAckHandle() {
+        return pendingAckHandle;
+    }
+
     public void syncBatchPositionBitSetForPendingAck(PositionImpl position) {
         this.pendingAckHandle.syncBatchPositionAckSetForTransaction(position);
     }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -382,6 +382,11 @@ private void initializeDispatchRateLimiterIfNeeded() {
         }
     }
 
+    @VisibleForTesting
+    public AtomicLong getPendingWriteOps() {
+        return pendingWriteOps;
+    }
+
     private PersistentSubscription createPersistentSubscription(String subscriptionName, ManagedCursor cursor,
             boolean replicated, Map<String, String> subscriptionProperties) {
         checkNotNull(compactedTopic);

pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java

@@ -21,6 +21,7 @@
 import static org.apache.bookkeeper.mledger.util.PositionAckSetUtil.andAckSet;
 import static org.apache.bookkeeper.mledger.util.PositionAckSetUtil.compareToWithAckSet;
 import static org.apache.bookkeeper.mledger.util.PositionAckSetUtil.isAckSetOverlap;
+import com.google.common.annotations.VisibleForTesting;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -1027,6 +1028,11 @@ public PositionInPendingAckStats checkPositionInPendingAckState(PositionImpl pos
         }
     }
 
+    @VisibleForTesting
+    public Map<PositionImpl, MutablePair<PositionImpl, Integer>> getIndividualAckPositions() {
+        return individualAckPositions;
+    }
+
     @Override
     public boolean checkIfPendingAckStoreInit() {
         return this.pendingAckStoreFuture != null && this.pendingAckStoreFuture.isDone();

pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java

@@ -37,10 +37,10 @@
 import org.apache.bookkeeper.net.CachedDNSToSwitchMapping;
 import org.apache.bookkeeper.stats.StatsLogger;
 import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.pulsar.bookie.rackawareness.BookieRackAffinityMapping;
 import org.apache.pulsar.metadata.api.MetadataStore;
 import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
-import org.powermock.reflect.Whitebox;
 import org.testng.annotations.Test;
 
 /**
@@ -281,7 +281,7 @@ public void testOpportunisticStripingConfiguration() {
     }
 
     @Test
-    public void testBookKeeperIoThreadsConfiguration() {
+    public void testBookKeeperIoThreadsConfiguration() throws Exception {
         BookKeeperClientFactoryImpl factory = new BookKeeperClientFactoryImpl();
         ServiceConfiguration conf = new ServiceConfiguration();
         assertEquals(factory.createBkClientConfiguration(mock(MetadataStoreExtended.class), conf)
@@ -292,11 +292,11 @@ public void testBookKeeperIoThreadsConfiguration() {
         EventLoopGroup eventLoopGroup = mock(EventLoopGroup.class);
         BookKeeper.Builder builder = factory.getBookKeeperBuilder(conf, eventLoopGroup,
                 mock(StatsLogger.class), mock(ClientConfiguration.class));
-        assertEquals(Whitebox.getInternalState(builder, "eventLoopGroup"), eventLoopGroup);
+        assertEquals(FieldUtils.readField(builder, "eventLoopGroup", true), eventLoopGroup);
         conf.setBookkeeperClientSeparatedIoThreadsEnabled(true);
         builder = factory.getBookKeeperBuilder(conf, eventLoopGroup,
                 mock(StatsLogger.class), mock(ClientConfiguration.class));
-        assertNull(Whitebox.getInternalState(builder, "eventLoopGroup"));
+        assertNull(FieldUtils.readField(builder, "eventLoopGroup", true));
     }
 
 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceCloseTest.java

@@ -24,9 +24,9 @@
 
 import java.util.concurrent.ScheduledFuture;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.loadbalance.LoadSheddingTask;
-import org.awaitility.reflect.WhiteboxImpl;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
@@ -62,18 +62,21 @@ protected PulsarService startBrokerWithoutAuthorization(ServiceConfiguration con
     @Test(timeOut = 30_000)
     public void closeInTimeTest() throws Exception {
         LoadSheddingTask task = pulsar.getLoadSheddingTask();
-        boolean isCancel = WhiteboxImpl.getInternalState(task, "isCancel");
-        assertFalse(isCancel);
-        ScheduledFuture<?> loadSheddingFuture = WhiteboxImpl.getInternalState(task, "future");
-        assertFalse(loadSheddingFuture.isCancelled());
+
+        {
+            assertFalse((boolean) FieldUtils.readField(task, "isCancel", true));
+            ScheduledFuture<?> loadSheddingFuture = (ScheduledFuture<?>) FieldUtils.readField(task, "future", true);
+            assertFalse(loadSheddingFuture.isCancelled());
+        }
 
         // The pulsar service is not used, so it should be closed gracefully in short time.
         pulsar.close();
 
-        isCancel = WhiteboxImpl.getInternalState(task, "isCancel");
-        assertTrue(isCancel);
-        loadSheddingFuture = WhiteboxImpl.getInternalState(task, "future");
-        assertTrue(loadSheddingFuture.isCancelled());
+        {
+            assertTrue((boolean) FieldUtils.readField(task, "isCancel", true));
+            ScheduledFuture<?> loadSheddingFuture = (ScheduledFuture<?>) FieldUtils.readField(task, "future", true);
+            assertTrue(loadSheddingFuture.isCancelled());
+        }
     }
 
 }

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

@@ -22,7 +22,6 @@
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
-import java.lang.reflect.Field;
 import java.util.Collection;
 import java.util.Date;
 import java.util.Map;
@@ -32,13 +31,11 @@
 import java.util.concurrent.TimeUnit;
 import javax.ws.rs.container.AsyncResponse;
 import javax.ws.rs.container.TimeoutHandler;
-import javax.ws.rs.core.UriInfo;
 import org.apache.pulsar.broker.admin.v2.PersistentTopics;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.authentication.AuthenticationDataHttps;
 import org.apache.pulsar.broker.service.Topic;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
-import org.apache.pulsar.broker.web.PulsarWebResource;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.MessageRoutingMode;
 import org.apache.pulsar.client.api.Producer;
@@ -51,26 +48,16 @@
 import org.awaitility.Awaitility;
 import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
-import org.testng.annotations.BeforeClass;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
 @Test(groups = "broker-admin")
 public class AdminApiGetLastMessageIdTest extends MockedPulsarServiceBaseTest {
 
-    private PersistentTopics persistentTopics;
-    private final String testTenant = "my-tenant";
-    private final String testLocalCluster = "use";
-    private final String testNamespace = "my-namespace";
-    protected Field uriField;
-    protected UriInfo uriInfo;
+    private static final String testTenant = "my-tenant";
+    private static final String testNamespace = "my-namespace";
 
-    @BeforeClass
-    public void initPersistentTopics() throws Exception {
-        uriField = PulsarWebResource.class.getDeclaredField("uri");
-        uriField.setAccessible(true);
-        uriInfo = mock(UriInfo.class);
-    }
+    private PersistentTopics persistentTopics;
 
     @Override
     @BeforeMethod

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

@@ -18,7 +18,6 @@
  */
 package org.apache.pulsar.broker.admin;
 
-import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -94,7 +93,6 @@
 import org.apache.zookeeper.KeeperException;
 import org.awaitility.Awaitility;
 import org.mockito.ArgumentCaptor;
-import org.powermock.reflect.Whitebox;
 import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeClass;
@@ -149,9 +147,8 @@ protected void setup() throws Exception {
 
         PulsarResources resources =
                 spy(new PulsarResources(pulsar.getLocalMetadataStore(), pulsar.getConfigurationMetadataStore()));
-        doReturn(spyWithClassAndConstructorArgs(TopicResources.class, pulsar.getLocalMetadataStore())).when(resources)
-                .getTopicResources();
-        Whitebox.setInternalState(pulsar, "pulsarResources", resources);
+        doReturn(spy(new TopicResources(pulsar.getLocalMetadataStore()))).when(resources).getTopicResources();
+        doReturn(resources).when(pulsar).getPulsarResources();
 
         admin.clusters().createCluster("use", ClusterData.builder().serviceUrl("http://broker-use.com:8080").build());
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl("http://broker-use.com:8080").build());

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

@@ -38,9 +38,9 @@
 import org.apache.pulsar.client.api.ProducerConsumerBase;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.LookupService;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
-import org.powermock.reflect.Whitebox;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
@@ -122,15 +122,14 @@ public void testPartitionedTopicAutoCreationForbiddenDuringNamespaceDeletion()
         });
 
 
-        LookupService original = Whitebox.getInternalState(pulsarClient, "lookup");
+        LookupService original = ((PulsarClientImpl) pulsarClient).getLookup();
         try {
 
             // we want to skip the "lookup" phase, because it is blocked by the HTTP API
             LookupService mockLookup = mock(LookupService.class);
-            Whitebox.setInternalState(pulsarClient, "lookup", mockLookup);
-            when(mockLookup.getPartitionedTopicMetadata(any())).thenAnswer(i -> {
-                return CompletableFuture.completedFuture(new PartitionedTopicMetadata(0));
-            });
+            ((PulsarClientImpl) pulsarClient).setLookup(mockLookup);
+            when(mockLookup.getPartitionedTopicMetadata(any())).thenAnswer(
+                    i -> CompletableFuture.completedFuture(new PartitionedTopicMetadata(0)));
             when(mockLookup.getBroker(any())).thenAnswer(i -> {
                 InetSocketAddress brokerAddress =
                         new InetSocketAddress(pulsar.getAdvertisedAddress(), pulsar.getBrokerListenPort().get());
@@ -139,20 +138,20 @@ public void testPartitionedTopicAutoCreationForbiddenDuringNamespaceDeletion()
 
             // Creating a producer and creating a Consumer may trigger automatic topic
             // creation, let's try to create a Producer and a Consumer
-            try (Producer<byte[]> producer = pulsarClient.newProducer()
+            try (Producer<byte[]> ignored = pulsarClient.newProducer()
                     .sendTimeout(1, TimeUnit.SECONDS)
                     .topic(topic)
-                    .create();) {
+                    .create()) {
             } catch (PulsarClientException.LookupException expected) {
                 String msg = "Namespace bundle for topic (%s) not served by this instance";
                 log.info("Expected error", expected);
                 assertTrue(expected.getMessage().contains(String.format(msg, topic)));
             }
 
-            try (Consumer<byte[]> consumer = pulsarClient.newConsumer()
+            try (Consumer<byte[]> ignored = pulsarClient.newConsumer()
                     .topic(topic)
                     .subscriptionName("test")
-                    .subscribe();) {
+                    .subscribe()) {
             } catch (PulsarClientException.LookupException expected) {
                 String msg = "Namespace bundle for topic (%s) not served by this instance";
                 log.info("Expected error", expected);
@@ -170,17 +169,16 @@ public void testPartitionedTopicAutoCreationForbiddenDuringNamespaceDeletion()
             admin.topics().getList(namespaceName).isEmpty();
 
             // create now the topic using auto creation
-            Whitebox.setInternalState(pulsarClient, "lookup", original);
-
-            try (Consumer<byte[]> consumer = pulsarClient.newConsumer()
+            ((PulsarClientImpl) pulsarClient).setLookup(original);
+            try (Consumer<byte[]> ignored = pulsarClient.newConsumer()
                     .topic(topic)
                     .subscriptionName("test")
-                    .subscribe();) {
+                    .subscribe()) {
             }
 
             admin.topics().getList(namespaceName).contains(topic);
         } finally {
-            Whitebox.setInternalState(pulsarClient, "lookup", original);
+            ((PulsarClientImpl) pulsarClient).setLookup(original);
         }
 
     }

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

@@ -49,6 +49,7 @@
 import org.apache.avro.io.JsonEncoder;
 import org.apache.avro.reflect.ReflectDatumWriter;
 import org.apache.avro.util.Utf8;
+import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.authentication.AuthenticationDataHttps;
@@ -88,7 +89,6 @@
 import org.mockito.ArgumentCaptor;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
-import org.powermock.reflect.Whitebox;
 import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
@@ -320,7 +320,7 @@ public void testLookUpWithRedirect() throws Exception {
         doReturn(false).when(topics).isRequestHttps();
         UriInfo uriInfo = mock(UriInfo.class);
         doReturn(requestPath).when(uriInfo).getRequestUri();
-        Whitebox.setInternalState(topics, "uri", uriInfo);
+        FieldUtils.writeField(topics, "uri", uriInfo, true);
         //do produce on another broker
         topics.setPulsar(pulsar2);
         AsyncResponse asyncResponse = mock(AsyncResponse.class);