[cleanup][misc] Fix all javac warnings in main sources

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java

@@ -26,6 +26,7 @@
 import com.google.common.collect.Multimap;
 import com.google.common.collect.Sets;
 import com.google.common.collect.TreeMultimap;
+import java.time.Duration;
 import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -238,7 +239,7 @@ public void initialize(final PulsarService pulsar) {
                 });
 
         int entryExpiryTime = (int) pulsar.getConfiguration().getLoadBalancerSheddingGracePeriodMinutes();
-        unloadedHotNamespaceCache = CacheBuilder.newBuilder().expireAfterWrite(entryExpiryTime, TimeUnit.MINUTES)
+        unloadedHotNamespaceCache = CacheBuilder.newBuilder().expireAfterWrite(Duration.ofMinutes(entryExpiryTime))
                 .build(new CacheLoader<String, Long>() {
                     @Override
                     public Long load(String key) throws Exception {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -578,7 +578,7 @@ private void startProtocolHandler(String protocol,
             bootstrap.option(ChannelOption.SO_REUSEADDR, true);
             bootstrap.childOption(ChannelOption.ALLOCATOR, PulsarByteBufAllocator.DEFAULT);
             bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
-            bootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,
+            bootstrap.childOption(ChannelOption.RECVBUF_ALLOCATOR,
                     new AdaptiveRecvByteBufAllocator(1024, 16 * 1024, 1 * 1024 * 1024));
             EventLoopUtil.enableTriggeredMode(bootstrap);
             DefaultThreadFactory defaultThreadFactory =
@@ -606,7 +606,7 @@ private ServerBootstrap defaultServerBootstrap() {
         bootstrap.childOption(ChannelOption.ALLOCATOR, PulsarByteBufAllocator.DEFAULT);
         bootstrap.group(acceptorGroup, workerGroup);
         bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
-        bootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,
+        bootstrap.childOption(ChannelOption.RECVBUF_ALLOCATOR,
             new AdaptiveRecvByteBufAllocator(1024, 16 * 1024, 1 * 1024 * 1024));
         bootstrap.channel(EventLoopUtil.getServerSocketChannelClass(workerGroup));
         EventLoopUtil.enableTriggeredMode(bootstrap);

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/LegacyAwareTopicPoliciesService.java

@@ -125,6 +125,7 @@ public CompletableFuture<Boolean> registerListenerAsync(TopicName topicName, Top
                 .thenCompose(service -> service.registerListenerAsync(topicName, listener));
     }
 
+    @Deprecated
     @Override
     public boolean registerListener(TopicName topicName, TopicPolicyListener listener) {
         throw new RuntimeException("should not be called");

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/MetadataStoreTopicPoliciesService.java

@@ -135,6 +135,7 @@ public CompletableFuture<Optional<TopicPolicies>> getTopicPoliciesAsync(TopicNam
                 .thenApply(policies -> policies.map(policy -> cloneWithScope(policy, global)));
     }
 
+    @Deprecated
     @Override
     public boolean registerListener(TopicName topicName, TopicPolicyListener listener) {
         listeners.compute(normalizeTopicName(topicName), (__, topicListeners) -> {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java

@@ -951,7 +951,7 @@ void cleanPoliciesCacheInitMap(@NonNull NamespaceName namespace) {
         // Complete the removed init future outside the compute() remapping function: completing it can run the
         // awaiting topic-load callbacks synchronously, and doing that while holding the ConcurrentHashMap bin lock
         // risks a recursive map update / deadlock (see #24977).
-        failPendingPolicyCacheInit(namespace, removedInitFuture.getValue());
+        failPendingPolicyCacheInit(namespace, removedInitFuture.get());
         if (readerFuture != null && !readerFuture.isCompletedExceptionally()) {
             readerFuture
                     .thenCompose(SystemTopicClient.Reader::closeAsync)
@@ -1152,6 +1152,7 @@ public CompletableFuture<Void> getPoliciesCacheInit(NamespaceName namespaceName)
         return policyCacheInitMap.get(namespaceName);
     }
 
+    @Deprecated
     @Override
     public boolean registerListener(TopicName topicName, TopicPolicyListener listener) {
         listeners.compute(topicName, (k, topicListeners) -> {

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

@@ -625,7 +625,8 @@ public CompletableFuture<Void> unloadSubscription(@NonNull String subName) {
      * @param subscriptionName the name of the subscription
      * @param cursor the cursor to use for the subscription
      * @param replicated the subscription replication flag
-     * @param subscriptionProperties the subscription properties
+     * @param subscriptionProperties the subscription properties. No longer used to build the subscription — the cursor
+     *                               already carries them — but retained so overriding implementations keep working.
      * @return the subscription instance
      */
     protected PersistentSubscription createPersistentSubscription(String subscriptionName, ManagedCursor cursor,
@@ -636,7 +637,7 @@ protected PersistentSubscription createPersistentSubscription(String subscriptio
             CompactedTopicImpl compactedTopic = pulsarTopicCompactionService.getCompactedTopic();
             return new PulsarCompactorSubscription(this, compactedTopic, subscriptionName, cursor);
         } else {
-            return new PersistentSubscription(this, subscriptionName, cursor, replicated, subscriptionProperties);
+            return new PersistentSubscription(this, subscriptionName, cursor, replicated);
         }
     }
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java

@@ -1264,10 +1264,10 @@ private CompletableFuture<Boolean> prunable(SegmentInfo seg, List<String> subs)
             // No subscribers ever attached / all unsubscribed → nothing left to drain.
             return CompletableFuture.completedFuture(true);
         }
-        CompletableFuture<Boolean>[] checks = subs.stream()
+        List<CompletableFuture<Boolean>> checks = subs.stream()
                 .map(sub -> isSegmentDrained(seg, sub))
-                .toArray(CompletableFuture[]::new);
-        return CompletableFuture.allOf(checks)
+                .toList();
+        return CompletableFuture.allOf(checks.toArray(CompletableFuture<?>[]::new))
                 .thenApply(__ -> {
                     for (CompletableFuture<Boolean> c : checks) {
                         if (!c.join()) {

pulsar-client-admin/build.gradle.kts

@@ -43,5 +43,9 @@ dependencies {
     implementation(libs.commons.lang3)
     implementation(libs.completable.futures)
 
+    // pulsar-client's configuration-data classes carry runtime-retained @Schema annotations, so javac needs
+    // the OpenAPI annotation types on the compile classpath to read their class files without warning.
+    compileOnly(libs.swagger.annotations)
+
     testImplementation(libs.wiremock)
 }

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java

@@ -637,6 +637,8 @@ public interface ConsumerBuilder<T> extends Cloneable {
      *
      * @param interceptors the list of interceptors to intercept the consumer created by this builder.
      */
+    // @SafeVarargs cannot be applied to an abstract method; implementations declare it on their final override.
+    @SuppressWarnings("unchecked")
     ConsumerBuilder<T> intercept(ConsumerInterceptor<T> ...interceptors);
 
     /**

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java

@@ -561,6 +561,8 @@ public interface ProducerBuilder<T> extends Cloneable {
      * @return the producer builder instance
      */
     @Deprecated
+    // @SafeVarargs cannot be applied to an abstract method; implementations declare it on their final override.
+    @SuppressWarnings("unchecked")
     ProducerBuilder<T> intercept(ProducerInterceptor<T> ... interceptors);
 
     /**

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ReaderBuilder.java

@@ -342,6 +342,8 @@ public interface ReaderBuilder<T> extends Cloneable {
      * @param interceptors the list of interceptors to intercept the reader created by this builder.
      * @return the reader builder instance
      */
+    // @SafeVarargs cannot be applied to an abstract method; implementations declare it on their final override.
+    @SuppressWarnings("unchecked")
     ReaderBuilder<T> intercept(ReaderInterceptor<T>... interceptors);
 
     /**

pulsar-client-v5/build.gradle.kts

@@ -35,6 +35,9 @@ dependencies {
     compileOnly(libs.protobuf.java)
     implementation(libs.netty.handler)
     implementation(libs.jackson.annotations)
+    // pulsar-client's configuration-data classes carry runtime-retained @Schema annotations, so javac needs
+    // the OpenAPI annotation types on the compile classpath to read their class files without warning.
+    compileOnly(libs.swagger.annotations)
     compileOnly(libs.lombok)
     annotationProcessor(libs.lombok)
     testImplementation(libs.testng)

pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/AuthenticationAdapter.java

@@ -110,6 +110,9 @@ public String authMethodName() {
         }
 
         @Override
+        // The v4 no-arg getAuthData() is deprecated in favour of the broker-host variant, but this bridge has to
+        // preserve the host-less semantics of the v5 no-arg authData() it implements.
+        @SuppressWarnings("deprecation")
         public AuthenticationData authData() throws PulsarClientException {
             try {
                 AuthenticationDataProvider v4Data = v4Auth.getAuthData();

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java

@@ -154,7 +154,7 @@ public enum State {
 
     private final LoadingCache<String, SchemaInfoProvider> schemaProviderLoadingCache =
             CacheBuilder.newBuilder().maximumSize(100000)
-                    .expireAfterAccess(30, TimeUnit.MINUTES)
+                    .expireAfterAccess(Duration.ofMinutes(30))
                     .build(new CacheLoader<String, SchemaInfoProvider>() {
 
                         @Override

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarServiceNameResolver.java

@@ -29,6 +29,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.stream.Collectors;
@@ -155,7 +156,7 @@ public synchronized void updateServiceUrl(String serviceUrl) throws InvalidServi
     private static int randomIndex(int numAddresses) {
         return numAddresses == 1
                 ?
-                0 : io.netty.util.internal.PlatformDependent.threadLocalRandom().nextInt(numAddresses);
+                0 : ThreadLocalRandom.current().nextInt(numAddresses);
     }
 
     /**

pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/MultiVersionSchemaInfoProvider.java

@@ -22,9 +22,9 @@
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 import lombok.CustomLog;
 import org.apache.pulsar.client.api.schema.SchemaInfoProvider;
 import org.apache.pulsar.client.impl.PulsarClientImpl;
@@ -44,7 +44,7 @@ public class MultiVersionSchemaInfoProvider implements SchemaInfoProvider {
 
     private final LoadingCache<BytesSchemaVersion, CompletableFuture<SchemaInfo>> cache = CacheBuilder.newBuilder()
         .maximumSize(100000)
-        .expireAfterAccess(30, TimeUnit.MINUTES)
+        .expireAfterAccess(Duration.ofMinutes(30))
         .build(new CacheLoader<BytesSchemaVersion, CompletableFuture<SchemaInfo>>() {
             @Override
             public CompletableFuture<SchemaInfo> load(BytesSchemaVersion schemaVersion) {

pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/AbstractMultiVersionReader.java

@@ -22,8 +22,8 @@
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
 import java.io.InputStream;
+import java.time.Duration;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 import lombok.CustomLog;
 import org.apache.avro.AvroTypeException;
 import org.apache.commons.codec.binary.Hex;
@@ -45,7 +45,7 @@ public abstract class AbstractMultiVersionReader<T> implements SchemaReader<T> {
     protected SchemaInfoProvider schemaInfoProvider;
 
     LoadingCache<BytesSchemaVersion, SchemaReader<T>> readerCache = CacheBuilder.newBuilder().maximumSize(100000)
-            .expireAfterAccess(30, TimeUnit.MINUTES).build(new CacheLoader<BytesSchemaVersion, SchemaReader<T>>() {
+            .expireAfterAccess(Duration.ofMinutes(30)).build(new CacheLoader<BytesSchemaVersion, SchemaReader<T>>() {
                 @Override
                 public SchemaReader<T> load(BytesSchemaVersion schemaVersion) {
                     return loadReader(schemaVersion);

pulsar-common/src/main/java/org/apache/pulsar/common/naming/NamespaceName.java

@@ -23,10 +23,10 @@
 import com.google.common.cache.LoadingCache;
 import com.google.common.util.concurrent.UncheckedExecutionException;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.time.Duration;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 
 /**
  * Parser of a value from the namespace field provided in configuration.
@@ -39,7 +39,7 @@ public class NamespaceName implements ServiceUnitId {
     private final String localName;
 
     private static final LoadingCache<String, NamespaceName> cache = CacheBuilder.newBuilder().maximumSize(100000)
-            .expireAfterAccess(30, TimeUnit.MINUTES).build(new CacheLoader<String, NamespaceName>() {
+            .expireAfterAccess(Duration.ofMinutes(30)).build(new CacheLoader<String, NamespaceName>() {
                 @Override
                 public NamespaceName load(String name) throws Exception {
                     return new NamespaceName(name);

pulsar-package-management/core/src/main/java/org/apache/pulsar/packages/management/core/common/PackageName.java

@@ -25,10 +25,10 @@
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
 import com.google.common.net.UrlEscapers;
+import java.time.Duration;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 
 /**
  * A package name has five parts, type, tenant, namespace, package-name, and version.
@@ -49,7 +49,7 @@ public class PackageName {
     private static final LoadingCache<String, PackageName> cache =
         CacheBuilder.newBuilder()
             .maximumSize(100000)
-            .expireAfterAccess(30, TimeUnit.MINUTES)
+            .expireAfterAccess(Duration.ofMinutes(30))
             .build(new CacheLoader<String, PackageName>() {
                 @Override
                 public PackageName load(String name) throws Exception {

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java

@@ -30,8 +30,6 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
-import io.netty.channel.epoll.EpollChannelOption;
-import io.netty.channel.epoll.EpollMode;
 import io.netty.channel.epoll.EpollSocketChannel;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.haproxy.HAProxyCommand;
@@ -160,9 +158,8 @@ public void connect(String brokerHostAndPort, InetSocketAddress targetBrokerAddr
         b.group(inboundChannel.eventLoop())
                 .channel(inboundChannel.getClass());
 
-        if (service.proxyZeroCopyModeEnabled && EpollSocketChannel.class.isAssignableFrom(inboundChannel.getClass())) {
-            b.option(EpollChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);
-        }
+        // Zero-copy (splice) requires the epoll transport to be level-triggered. Netty always uses
+        // level-triggered mode since 4.2, so no channel option needs to be set for it anymore.
 
         b.handler(new ChannelInitializer<SocketChannel>() {
             @Override

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java

@@ -255,7 +255,7 @@ public void start() throws Exception {
         bootstrap.childOption(ChannelOption.ALLOCATOR, PulsarByteBufAllocator.DEFAULT);
         bootstrap.group(acceptorGroup, workerGroup);
         bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
-        bootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,
+        bootstrap.childOption(ChannelOption.RECVBUF_ALLOCATOR,
                 new AdaptiveRecvByteBufAllocator(1024, 16 * 1024, 1 * 1024 * 1024));
 
         Class<? extends ServerSocketChannel> serverSocketChannelClass =
@@ -361,7 +361,7 @@ private void startProxyExtension(String extensionName,
             bootstrap.option(ChannelOption.SO_REUSEADDR, true);
             bootstrap.childOption(ChannelOption.ALLOCATOR, PulsarByteBufAllocator.DEFAULT);
             bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
-            bootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,
+            bootstrap.childOption(ChannelOption.RECVBUF_ALLOCATOR,
                     new AdaptiveRecvByteBufAllocator(1024, 16 * 1024, 1 * 1024 * 1024));
 
             EventLoopUtil.enableTriggeredMode(bootstrap);

pulsar-testclient/build.gradle.kts

@@ -54,6 +54,10 @@ dependencies {
     implementation(libs.jetty.util)
     implementation(libs.opentelemetry.sdk.extension.autoconfigure)
 
+    // pulsar-client's configuration-data classes carry runtime-retained @Schema annotations, so javac needs
+    // the OpenAPI annotation types on the compile classpath to read their class files without warning.
+    compileOnly(libs.swagger.annotations)
+
     testImplementation(project(":pulsar-broker"))
     testImplementation(project(path = ":pulsar-broker", configuration = "testJar"))
 }

pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java

@@ -262,7 +262,7 @@ public void onWebSocketError(Throwable cause) {
     }
 
     @Override
-    public void onWebSocketClose(int statusCode, String reason) {
+    public void onWebSocketClose(int statusCode, String reason, Callback callback) {
         log.info()
                 .attr("remoteAddress", getSession().getRemoteSocketAddress())
                 .attr("topic", topic)
@@ -279,6 +279,7 @@ public void onWebSocketClose(int statusCode, String reason) {
                     .exception(e)
                     .log("Failed to close handler for topic");
         }
+        callback.succeed();
     }
 
     public void close(WebSocketError error) {

pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java

@@ -27,6 +27,7 @@
 import com.google.common.cache.CacheBuilder;
 import jakarta.servlet.http.HttpServletRequest;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.Base64;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -96,7 +97,7 @@ public class ConsumerHandler extends AbstractWebSocketHandler {
     // Make sure use the same BatchMessageIdImpl to acknowledge the batch message, otherwise the BatchMessageAcker
     // of the BatchMessageIdImpl will not complete.
     private Cache<String, MessageId> messageIdCache = CacheBuilder.newBuilder()
-            .expireAfterWrite(1, TimeUnit.HOURS)
+            .expireAfterWrite(Duration.ofHours(1))
             .build();
 
     public ConsumerHandler(WebSocketService service, HttpServletRequest request, JettyServerUpgradeResponse response) {

pulsar-websocket/src/test/java/org/apache/pulsar/websocket/AbstractWebSocketHandlerTest.java

@@ -57,6 +57,7 @@
 import org.apache.pulsar.websocket.service.WebSocketProxyConfiguration;
 import org.eclipse.jetty.ee10.websocket.server.JettyServerUpgradeResponse;
 import org.eclipse.jetty.http.HttpStatus;
+import org.eclipse.jetty.websocket.api.Callback;
 import org.eclipse.jetty.websocket.api.Session;
 import org.mockito.Answers;
 import org.mockito.Mock;
@@ -395,7 +396,8 @@ public void testPingFuture() throws IOException {
         assertNotNull(pingFuture);
         assertFalse(pingFuture.isDone());
 
-        webSocketHandler.onWebSocketClose(HttpStatus.INTERNAL_SERVER_ERROR_500, "INTERNAL_SERVER_ERROR_500");
+        webSocketHandler.onWebSocketClose(HttpStatus.INTERNAL_SERVER_ERROR_500, "INTERNAL_SERVER_ERROR_500",
+                Callback.NOOP);
         assertTrue(pingFuture.isDone());