You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/05/05 13:07:03 UTC
(pulsar) 06/08: [improve][misc] Migrate from multiple nullness annotation libraries to JSpecify annotations (#24239)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 62c505da3d42a244f45f95bb78717c17281fc4ad
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat May 3 14:47:06 2025 +0300
[improve][misc] Migrate from multiple nullness annotation libraries to JSpecify annotations (#24239)
(cherry picked from commit 9a1124516ed46cf59cba39ffa7b12a0caf7b9d80)
---
.../src/main/resources/pulsar/checkstyle.xml | 16 ++++++++++
distribution/server/src/assemble/LICENSE.bin.txt | 2 ++
distribution/shell/src/assemble/LICENSE.bin.txt | 1 +
.../bookkeeper/mledger/impl/AckSetState.java | 2 +-
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 2 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 2 +-
pom.xml | 5 ++++
.../oidc/OpenIDProviderMetadataCache.java | 6 ++--
pulsar-broker/pom.xml | 4 +++
.../pulsar/broker/admin/impl/NamespacesBase.java | 6 ++--
.../broker/admin/impl/PersistentTopicsBase.java | 4 +--
.../pulsar/broker/admin/v3/Transactions.java | 2 +-
.../bucket/BookkeeperBucketSnapshotStorage.java | 4 +--
.../broker/namespace/NamespaceEphemeralData.java | 4 +--
.../pulsar/broker/namespace/NamespaceService.java | 2 +-
.../broker/service/AbstractBaseDispatcher.java | 2 +-
.../pulsar/broker/service/AbstractTopic.java | 4 +--
.../pulsar/broker/service/BrokerService.java | 12 ++++----
.../service/ConsumerHashAssignmentsSnapshot.java | 4 +--
.../pulsar/broker/service/EntryAndMetadata.java | 2 +-
.../SystemTopicBasedTopicPoliciesService.java | 8 ++---
.../persistent/PersistentMessageExpiryMonitor.java | 2 +-
.../broker/service/persistent/PersistentTopic.java | 6 ++--
.../broker/service/persistent/SystemTopic.java | 4 +--
.../service/schema/BookkeeperSchemaStorage.java | 34 +++++++++++-----------
.../schema/BookkeeperSchemaStorageFactory.java | 4 +--
.../service/schema/SchemaRegistryServiceImpl.java | 8 ++---
.../service/schema/SchemaStorageFactory.java | 4 +--
.../pulsar/broker/web/PulsarWebResource.java | 4 +--
.../pulsar/compaction/CompactedTopicImpl.java | 2 +-
.../pulsar/compaction/CompactedTopicUtils.java | 2 +-
.../compaction/CompactionServiceFactory.java | 6 ++--
.../compaction/PulsarCompactionServiceFactory.java | 8 ++---
.../compaction/PulsarTopicCompactionService.java | 4 +--
.../pulsar/compaction/TopicCompactionService.java | 4 +--
.../pulsar/broker/MultiBrokerTestZKBaseTest.java | 4 +--
.../ManagedLedgerInterceptorImplTest.java | 3 +-
.../broker/testcontext/PulsarTestContext.java | 4 +--
.../MultiBrokerMetadataConsistencyTest.java | 4 +--
.../pulsar/compaction/CompactedTopicImplTest.java | 4 +--
pulsar-client/pom.xml | 5 ++++
.../client/impl/ControlledClusterFailover.java | 2 +-
.../client/impl/MultiTopicsConsumerImpl.java | 2 +-
.../PersistentAcknowledgmentsGroupingTracker.java | 2 +-
.../pulsar/client/impl/PulsarClientImpl.java | 2 +-
pulsar-common/pom.xml | 5 ++++
.../org/apache/pulsar/common/util/FutureUtil.java | 4 +--
.../pulsar/common/util/URIPreconditions.java | 16 +++++-----
.../collections/GrowableArrayBlockingQueue.java | 2 +-
.../instance/JavaInstanceRunnableTest.java | 4 +--
.../functions/instance/JavaInstanceDepsTest.java | 1 +
.../pulsar/io/azuredataexplorer/ADXSink.java | 8 ++---
.../io/azuredataexplorer/ADXSinkConfigTest.java | 6 ++--
pulsar-io/core/pom.xml | 4 +++
.../server/ProxyWithExtensibleLoadManagerTest.java | 4 +--
55 files changed, 157 insertions(+), 115 deletions(-)
diff --git a/buildtools/src/main/resources/pulsar/checkstyle.xml b/buildtools/src/main/resources/pulsar/checkstyle.xml
index 14808cf8663..12126ab6154 100644
--- a/buildtools/src/main/resources/pulsar/checkstyle.xml
+++ b/buildtools/src/main/resources/pulsar/checkstyle.xml
@@ -140,6 +140,22 @@ page at http://checkstyle.sourceforge.net/config.html -->
value="autovalue.shaded, avro.shaded, bk-shade, com.google.api.client.repackaged, com.google.appengine.repackaged, org.apache.curator.shaded, org.testcontainers.shaded, org.junit" />
</module>
+ <module name="IllegalImport">
+ <property name="illegalPkgs"
+ value="org.jetbrains.annotations, org.checkerframework.checker.nullness.qual" />
+ <property name="illegalClasses"
+ value="javax.annotation.Nullable, javax.annotation.Nonnull" />
+ <message key="import.illegal"
+ value="{0}; Use org.jspecify.annotations.Nullable and org.jspecify.annotations.NonNull instead."/>
+ </module>
+
+ <module name="IllegalImport">
+ <property name="illegalClasses"
+ value="javax.validation.constraints.Null, javax.validation.constraints.NotNull" />
+ <message key="import.illegal"
+ value="{0}; Use org.jspecify.annotations.Nullable and org.jspecify.annotations.NonNull for nullness information. Check Jakarta Bean Validation API if you are looking for user input validation with annotations."/>
+ </module>
+
<module name="RedundantModifier">
<!-- Checks for redundant modifiers on various symbol definitions.
See: http://checkstyle.sourceforge.net/config_modifier.html#RedundantModifier
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index 3e479c91e2b..fd9c2467bb4 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -539,6 +539,8 @@ The Apache Software License, Version 2.0
- io.opentelemetry.semconv-opentelemetry-semconv-1.29.0-alpha.jar
* Spotify completable-futures
- com.spotify-completable-futures-0.3.6.jar
+ * JSpecify
+ - org.jspecify-jspecify-1.0.0.jar
BSD 3-clause "New" or "Revised" License
* Google auth library
diff --git a/distribution/shell/src/assemble/LICENSE.bin.txt b/distribution/shell/src/assemble/LICENSE.bin.txt
index b6b544541b2..2773ff2eced 100644
--- a/distribution/shell/src/assemble/LICENSE.bin.txt
+++ b/distribution/shell/src/assemble/LICENSE.bin.txt
@@ -420,6 +420,7 @@ The Apache Software License, Version 2.0
* Spotify completable-futures -- completable-futures-0.3.6.jar
* RoaringBitmap -- RoaringBitmap-1.2.0.jar
* Fastutil -- fastutil-8.5.14.jar
+ * JSpecify -- jspecify-1.0.0.jar
BSD 3-clause "New" or "Revised" License
* JSR305 -- jsr305-3.0.2.jar -- ../licenses/LICENSE-JSR305.txt
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/AckSetState.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/AckSetState.java
index 6e9bf78b14c..03ff50c1c7f 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/AckSetState.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/AckSetState.java
@@ -18,7 +18,7 @@
*/
package org.apache.bookkeeper.mledger.impl;
-import javax.annotation.Nullable;
+import org.jspecify.annotations.Nullable;
/**
* Interface to manage the ackSet state attached to a position.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index ca221491c5e..f4a64d267fd 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -63,7 +63,6 @@ import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
-import javax.annotation.Nullable;
import lombok.Getter;
import org.apache.bookkeeper.client.AsyncCallback.CloseCallback;
import org.apache.bookkeeper.client.AsyncCallback.OpenCallback;
@@ -113,6 +112,7 @@ import org.apache.pulsar.common.util.collections.LongPairRangeSet;
import org.apache.pulsar.common.util.collections.LongPairRangeSet.LongPairConsumer;
import org.apache.pulsar.common.util.collections.LongPairRangeSet.RangeBoundConsumer;
import org.apache.pulsar.metadata.api.Stat;
+import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index c0884ee70c2..9545c8b138b 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -68,7 +68,6 @@ import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collectors;
-import javax.annotation.Nullable;
import lombok.Getter;
import org.apache.bookkeeper.client.AsyncCallback;
import org.apache.bookkeeper.client.AsyncCallback.CreateCallback;
@@ -149,6 +148,7 @@ import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.LazyLoadableValue;
import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap;
import org.apache.pulsar.metadata.api.Stat;
+import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/pom.xml b/pom.xml
index 5ad7056a516..68f51d2fab9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1639,6 +1639,11 @@ flexible messaging model and an intuitive client API.</description>
<artifactId>checker-qual</artifactId>
<version>${checkerframework.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.jspecify</groupId>
+ <artifactId>jspecify</artifactId>
+ <version>1.0.0</version>
+ </dependency>
<dependency>
<groupId>io.rest-assured</groupId>
diff --git a/pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/OpenIDProviderMetadataCache.java b/pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/OpenIDProviderMetadataCache.java
index cffa52b00aa..ebf8168b494 100644
--- a/pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/OpenIDProviderMetadataCache.java
+++ b/pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/OpenIDProviderMetadataCache.java
@@ -40,11 +40,11 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
-import javax.annotation.Nonnull;
import javax.naming.AuthenticationException;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationProvider;
import org.asynchttpclient.AsyncHttpClient;
+import org.jspecify.annotations.NonNull;
/**
* Class used to cache metadata responses from OpenID Providers.
@@ -94,7 +94,7 @@ class OpenIDProviderMetadataCache {
* @return the {@link OpenIDProviderMetadata} for the given issuer. Fail the completable future with
* AuthenticationException if any exceptions occur while retrieving the metadata.
*/
- CompletableFuture<OpenIDProviderMetadata> getOpenIDProviderMetadataForIssuer(@Nonnull String issuer) {
+ CompletableFuture<OpenIDProviderMetadata> getOpenIDProviderMetadataForIssuer(@NonNull String issuer) {
return cache.get(Optional.of(issuer));
}
@@ -221,7 +221,7 @@ class OpenIDProviderMetadataCache {
* @param isK8s - whether the issuer is represented by the Kubernetes API server. This affects error reporting.
* @throws AuthenticationException if the issuer does not exactly match the metadata issuer
*/
- private void verifyIssuer(@Nonnull String issuer, OpenIDProviderMetadata metadata,
+ private void verifyIssuer(@NonNull String issuer, OpenIDProviderMetadata metadata,
boolean isK8s) throws AuthenticationException {
if (!issuer.equals(metadata.getIssuer())) {
if (isK8s) {
diff --git a/pulsar-broker/pom.xml b/pulsar-broker/pom.xml
index e49fcb60565..770909d0247 100644
--- a/pulsar-broker/pom.xml
+++ b/pulsar-broker/pom.xml
@@ -319,6 +319,10 @@
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.jspecify</groupId>
+ <artifactId>jspecify</artifactId>
+ </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
index 61db138e60c..b6c577923aa 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -43,7 +43,6 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import javax.annotation.Nonnull;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.core.Response;
@@ -118,6 +117,7 @@ import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException;
import org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException;
import org.apache.zookeeper.KeeperException;
+import org.jspecify.annotations.NonNull;
@Slf4j
public abstract class NamespacesBase extends AdminResource {
@@ -211,13 +211,13 @@ public abstract class NamespacesBase extends AdminResource {
* Delete the namespace and retry to resolve some topics that were not created successfully(in metadata)
* during the deletion.
*/
- protected @Nonnull CompletableFuture<Void> internalDeleteNamespaceAsync(boolean force) {
+ protected @NonNull CompletableFuture<Void> internalDeleteNamespaceAsync(boolean force) {
final CompletableFuture<Void> future = new CompletableFuture<>();
internalRetryableDeleteNamespaceAsync0(force, 5, future);
return future;
}
private void internalRetryableDeleteNamespaceAsync0(boolean force, int retryTimes,
- @Nonnull CompletableFuture<Void> callback) {
+ @NonNull CompletableFuture<Void> callback) {
precheckWhenDeleteNamespace(namespaceName, force)
.thenCompose(policies -> {
final CompletableFuture<List<String>> topicsFuture;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index de99dd44626..169a1b750c2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -49,7 +49,6 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
-import javax.annotation.Nonnull;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.core.Response;
@@ -151,6 +150,7 @@ import org.apache.pulsar.common.util.collections.BitSetRecyclable;
import org.apache.pulsar.compaction.Compactor;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.transaction.coordinator.TransactionCoordinatorID;
+import org.jspecify.annotations.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -357,7 +357,7 @@ public class PersistentTopicsBase extends AdminResource {
* @exception RestException Unprocessable entity, status code: 422. throw it when some pre-check failed.
* @exception RestException Internal Server Error, status code: 500. throw it when get unknown Exception
*/
- protected @Nonnull CompletableFuture<Void> internalUpdatePartitionedTopicAsync(int expectPartitions,
+ protected @NonNull CompletableFuture<Void> internalUpdatePartitionedTopicAsync(int expectPartitions,
boolean updateLocal,
boolean force) {
PulsarService pulsarService = pulsar();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java
index 089ec530692..ab1787d2b2c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java
@@ -56,7 +56,7 @@ import org.apache.pulsar.common.policies.data.TransactionPendingAckInternalStats
import org.apache.pulsar.common.policies.data.TransactionPendingAckStats;
import org.apache.pulsar.common.stats.PositionInPendingAckStats;
import org.apache.pulsar.common.util.FutureUtil;
-import org.jetbrains.annotations.Nullable;
+import org.jspecify.annotations.Nullable;
@Path("/transactions")
@Produces(MediaType.APPLICATION_JSON)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
index 0d90e5e1d98..ba37092e88d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
@@ -28,7 +28,6 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
-import javax.validation.constraints.NotNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
@@ -41,6 +40,7 @@ import org.apache.pulsar.broker.delayed.proto.SnapshotMetadata;
import org.apache.pulsar.broker.delayed.proto.SnapshotSegment;
import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
import org.apache.pulsar.common.util.FutureUtil;
+import org.jspecify.annotations.NonNull;
@Slf4j
public class BookkeeperBucketSnapshotStorage implements BucketSnapshotStorage {
@@ -165,7 +165,7 @@ public class BookkeeperBucketSnapshotStorage implements BucketSnapshotStorage {
return snapshotMetadataList;
}
- @NotNull
+ @NonNull
private CompletableFuture<LedgerHandle> createLedger(String bucketKey, String topicName, String cursorName) {
CompletableFuture<LedgerHandle> future = new CompletableFuture<>();
Map<String, byte[]> metadata = LedgerMetadataUtils.buildMetadataForDelayedIndexBucket(bucketKey,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceEphemeralData.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceEphemeralData.java
index 1ffc03786d5..825942d0de1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceEphemeralData.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceEphemeralData.java
@@ -21,10 +21,10 @@ package org.apache.pulsar.broker.namespace;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-import javax.validation.constraints.NotNull;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.pulsar.policies.data.loadbalancer.AdvertisedListener;
+import org.jspecify.annotations.NonNull;
@Data
@NoArgsConstructor
@@ -55,7 +55,7 @@ public class NamespaceEphemeralData {
}
}
- @NotNull
+ @NonNull
public Map<String, AdvertisedListener> getAdvertisedListeners() {
if (this.advertisedListeners == null) {
return Collections.emptyMap();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index b2ee299bb03..032a868f2b3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -52,7 +52,6 @@ import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
-import javax.annotation.Nullable;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.ListUtils;
@@ -114,6 +113,7 @@ import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.opentelemetry.annotations.PulsarDeprecatedMetric;
import org.apache.pulsar.policies.data.loadbalancer.AdvertisedListener;
import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
+import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
index b7b300685ad..f074f234b87 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
@@ -53,7 +53,7 @@ import org.apache.pulsar.common.api.proto.ReplicatedSubscriptionsSnapshot;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.protocol.Markers;
import org.apache.pulsar.compaction.Compactor;
-import org.checkerframework.checker.nullness.qual.Nullable;
+import org.jspecify.annotations.Nullable;
@Slf4j
public abstract class AbstractBaseDispatcher extends EntryFilterSupport implements Dispatcher {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index f6397f21962..617d5a6b124 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -46,7 +46,6 @@ import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.ToLongFunction;
-import javax.annotation.Nonnull;
import lombok.Getter;
import lombok.Setter;
import org.apache.bookkeeper.mledger.util.StatsBuckets;
@@ -88,6 +87,7 @@ import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl;
import org.apache.pulsar.common.protocol.schema.SchemaData;
import org.apache.pulsar.common.protocol.schema.SchemaVersion;
import org.apache.pulsar.common.util.FutureUtil;
+import org.jspecify.annotations.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -1168,7 +1168,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener {
updateResourceGroupLimiter(policies);
}
- public void updateResourceGroupLimiter(@Nonnull Policies namespacePolicies) {
+ public void updateResourceGroupLimiter(@NonNull Policies namespacePolicies) {
requireNonNull(namespacePolicies);
// attach the resource-group level rate limiters, if set
String rgName = namespacePolicies.resource_group_name;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index b1a6cec153c..75ddbf9c904 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -80,8 +80,6 @@ import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Consumer;
import java.util.function.Predicate;
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
import lombok.AccessLevel;
import lombok.AllArgsConstructor;
import lombok.Getter;
@@ -196,6 +194,8 @@ import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes.ConnectionRateLim
import org.apache.pulsar.opentelemetry.annotations.PulsarDeprecatedMetric;
import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
import org.apache.pulsar.transaction.coordinator.TransactionMetadataStore;
+import org.jspecify.annotations.NonNull;
+import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -1202,7 +1202,7 @@ public class BrokerService implements Closeable {
}
}
- private CompletableFuture<Optional<TopicPolicies>> getTopicPoliciesBypassSystemTopic(@Nonnull TopicName topicName) {
+ private CompletableFuture<Optional<TopicPolicies>> getTopicPoliciesBypassSystemTopic(@NonNull TopicName topicName) {
if (ExtensibleLoadManagerImpl.isInternalTopic(topicName.toString())) {
return CompletableFuture.completedFuture(Optional.empty());
}
@@ -1935,14 +1935,14 @@ public class BrokerService implements Closeable {
return result;
}
- public CompletableFuture<ManagedLedgerConfig> getManagedLedgerConfig(@Nonnull TopicName topicName) {
+ public CompletableFuture<ManagedLedgerConfig> getManagedLedgerConfig(@NonNull TopicName topicName) {
final CompletableFuture<Optional<TopicPolicies>> topicPoliciesFuture =
getTopicPoliciesBypassSystemTopic(topicName);
return topicPoliciesFuture.thenCompose(optionalTopicPolicies ->
getManagedLedgerConfig(topicName, optionalTopicPolicies.orElse(null)));
}
- private CompletableFuture<ManagedLedgerConfig> getManagedLedgerConfig(@Nonnull TopicName topicName,
+ private CompletableFuture<ManagedLedgerConfig> getManagedLedgerConfig(@NonNull TopicName topicName,
@Nullable TopicPolicies topicPolicies) {
requireNonNull(topicName);
NamespaceName namespace = topicName.getNamespaceObject();
@@ -3595,7 +3595,7 @@ public class BrokerService implements Closeable {
return null;
}
- public @Nonnull CompletableFuture<Boolean> isAllowAutoSubscriptionCreationAsync(@Nonnull TopicName tpName) {
+ public @NonNull CompletableFuture<Boolean> isAllowAutoSubscriptionCreationAsync(@NonNull TopicName tpName) {
requireNonNull(tpName);
// Policies priority: topic level -> namespace level -> broker level
if (ExtensibleLoadManagerImpl.isInternalTopic(tpName.toString())) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsumerHashAssignmentsSnapshot.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsumerHashAssignmentsSnapshot.java
index b4add79294c..18ea927c642 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsumerHashAssignmentsSnapshot.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsumerHashAssignmentsSnapshot.java
@@ -32,7 +32,7 @@ import lombok.EqualsAndHashCode;
import lombok.ToString;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.client.api.Range;
-import org.jetbrains.annotations.NotNull;
+import org.jspecify.annotations.NonNull;
/**
* Represents the hash ranges assigned to each consumer in a {@link StickyKeyConsumerSelector} at a point in time.
@@ -86,7 +86,7 @@ public class ConsumerHashAssignmentsSnapshot {
return cachedRangesByConsumer;
}
- private @NotNull Map<Consumer, List<Range>> internalGetRangesByConsumer() {
+ private @NonNull Map<Consumer, List<Range>> internalGetRangesByConsumer() {
Map<Consumer, SortedSet<Range>> rangesByConsumer = new IdentityHashMap<>();
hashRangeAssignments.forEach(entry -> {
Range range = entry.range();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryAndMetadata.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryAndMetadata.java
index e4870bf251e..cf1802e3853 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryAndMetadata.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryAndMetadata.java
@@ -21,12 +21,12 @@ package org.apache.pulsar.broker.service;
import com.google.common.annotations.VisibleForTesting;
import io.netty.buffer.ByteBuf;
import java.util.function.ToIntFunction;
-import javax.annotation.Nullable;
import lombok.Getter;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.Position;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.protocol.Commands;
+import org.jspecify.annotations.Nullable;
public class EntryAndMetadata implements Entry {
private static final int STICKY_KEY_HASH_NOT_INITIALIZED = -1;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
index 7bf6119de34..2000c9a3099 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -34,7 +34,6 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
-import javax.annotation.Nonnull;
import org.apache.commons.lang3.concurrent.ConcurrentInitializer;
import org.apache.commons.lang3.concurrent.LazyInitializer;
import org.apache.commons.lang3.mutable.Mutable;
@@ -61,6 +60,7 @@ import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.TopicPolicies;
import org.apache.pulsar.common.util.FutureUtil;
+import org.jspecify.annotations.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -331,7 +331,7 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
}
@VisibleForTesting
- @Nonnull CompletableFuture<Boolean> prepareInitPoliciesCacheAsync(@Nonnull NamespaceName namespace) {
+ @NonNull CompletableFuture<Boolean> prepareInitPoliciesCacheAsync(@NonNull NamespaceName namespace) {
requireNonNull(namespace);
if (closed.get()) {
return CompletableFuture.completedFuture(false);
@@ -485,11 +485,11 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
});
}
- private void cleanCacheAndCloseReader(@Nonnull NamespaceName namespace, boolean cleanOwnedBundlesCount) {
+ private void cleanCacheAndCloseReader(@NonNull NamespaceName namespace, boolean cleanOwnedBundlesCount) {
cleanCacheAndCloseReader(namespace, cleanOwnedBundlesCount, false);
}
- private void cleanCacheAndCloseReader(@Nonnull NamespaceName namespace, boolean cleanOwnedBundlesCount,
+ private void cleanCacheAndCloseReader(@NonNull NamespaceName namespace, boolean cleanOwnedBundlesCount,
boolean cleanWriterCache) {
if (cleanWriterCache) {
writerCaches.synchronous().invalidate(namespace);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
index 4cd696b0eb7..6404e56d59b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
@@ -24,7 +24,6 @@ import java.util.Optional;
import java.util.SortedMap;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.LongAdder;
-import javax.annotation.Nullable;
import org.apache.bookkeeper.mledger.AsyncCallbacks.FindEntryCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.MarkDeleteCallback;
import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -40,6 +39,7 @@ import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.stats.Rate;
+import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 5dd2909def8..02d6bdd12bd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -57,7 +57,6 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.BiFunction;
-import javax.annotation.Nonnull;
import lombok.Getter;
import lombok.Value;
import org.apache.bookkeeper.client.BKException.BKNoSuchLedgerExistsException;
@@ -199,6 +198,7 @@ import org.apache.pulsar.compaction.TopicCompactionService;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
import org.apache.pulsar.utils.StatsOutputStream;
+import org.jspecify.annotations.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -547,7 +547,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
* @throws SubscriptionConflictUnloadException Conflict topic-close, topic-delete, another-subscribe-unload,
* cannot unload subscription now
*/
- public CompletableFuture<Void> unloadSubscription(@Nonnull String subName) {
+ public CompletableFuture<Void> unloadSubscription(@NonNull String subName) {
final PersistentSubscription sub = subscriptions.get(subName);
if (sub == null) {
return CompletableFuture.failedFuture(
@@ -3436,7 +3436,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
@Override
- public CompletableFuture<Void> onPoliciesUpdate(@Nonnull Policies data) {
+ public CompletableFuture<Void> onPoliciesUpdate(@NonNull Policies data) {
requireNonNull(data);
if (log.isDebugEnabled()) {
log.debug("[{}] isEncryptionRequired changes: {} -> {}", topic, isEncryptionRequired,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SystemTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SystemTopic.java
index a26255c9f8b..98711dab9b0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SystemTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SystemTopic.java
@@ -21,7 +21,6 @@ package org.apache.pulsar.broker.service.persistent;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import javax.annotation.Nonnull;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.namespace.NamespaceService;
@@ -33,6 +32,7 @@ import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.EntryFilters;
import org.apache.pulsar.common.policies.data.Policies;
+import org.jspecify.annotations.NonNull;
public class SystemTopic extends PersistentTopic {
@@ -123,7 +123,7 @@ public class SystemTopic extends PersistentTopic {
}
@Override
- public void updateResourceGroupLimiter(@Nonnull Policies namespacePolicies) {
+ public void updateResourceGroupLimiter(@NonNull Policies namespacePolicies) {
// nothing todo.
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
index f68cdd6473e..b931239a32c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
@@ -42,7 +42,6 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.function.Function;
import java.util.stream.Collectors;
-import javax.validation.constraints.NotNull;
import org.apache.bookkeeper.client.AsyncCallback;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
@@ -67,6 +66,7 @@ import org.apache.pulsar.metadata.api.MetadataSerde;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.Stat;
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+import org.jspecify.annotations.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -237,7 +237,7 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
return delete(key, false);
}
- @NotNull
+ @NonNull
private CompletableFuture<StoredSchema> getSchema(String schemaId) {
// There's already a schema read operation in progress. Just piggyback on that
return readSchemaOperations.computeIfAbsent(schemaId, key -> {
@@ -285,7 +285,7 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
}
}
- @NotNull
+ @NonNull
private CompletableFuture<StoredSchema> getSchema(String schemaId, long version) {
if (log.isDebugEnabled()) {
log.debug("[{}] Get schema - version: {}", schemaId, version);
@@ -315,7 +315,7 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
});
}
- @NotNull
+ @NonNull
private CompletableFuture<Long> putSchema(String schemaId, byte[] data, byte[] hash) {
return getSchemaLocator(getSchemaPath(schemaId)).thenCompose(optLocatorEntry ->
putSchema(schemaId, data, hash, optLocatorEntry));
@@ -368,7 +368,7 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
});
}
- @NotNull
+ @NonNull
private CompletableFuture<Long> deleteSchema(String schemaId, boolean forcefully) {
return (forcefully ? CompletableFuture.completedFuture(null)
: ignoreUnrecoverableBKException(getSchema(schemaId))).thenCompose(schemaAndVersion -> {
@@ -428,12 +428,12 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
});
}
- @NotNull
+ @NonNull
private static String getSchemaPath(String schemaId) {
return SchemaPath + "/" + schemaId;
}
- @NotNull
+ @NonNull
private CompletableFuture<SchemaStorageFormat.PositionInfo> addNewSchemaEntryToStore(
String schemaId,
List<SchemaStorageFormat.IndexEntry> index,
@@ -450,7 +450,7 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
});
}
- @NotNull
+ @NonNull
private CompletableFuture<Long> updateSchemaLocator(
String schemaId,
LocatorEntry locatorEntry,
@@ -494,7 +494,7 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
});
}
- @NotNull
+ @NonNull
private CompletableFuture<SchemaStorageFormat.SchemaEntry> findSchemaEntryByVersion(
List<SchemaStorageFormat.IndexEntry> index,
long version
@@ -521,7 +521,7 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
return completedFuture(null);
}
- @NotNull
+ @NonNull
private CompletableFuture<SchemaStorageFormat.SchemaEntry> readSchemaEntry(
SchemaStorageFormat.PositionInfo position
) {
@@ -538,19 +538,19 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
).thenCompose(Functions::parseSchemaEntry);
}
- @NotNull
+ @NonNull
private CompletableFuture<Void> updateSchemaLocator(String id,
SchemaStorageFormat.SchemaLocator schema, long version) {
return store.put(id, schema.toByteArray(), Optional.of(version)).thenApply(__ -> null);
}
- @NotNull
+ @NonNull
private CompletableFuture<LocatorEntry> createSchemaLocator(String id, SchemaStorageFormat.SchemaLocator locator) {
return store.put(id, locator.toByteArray(), Optional.of(-1L))
.thenApply(stat -> new LocatorEntry(locator, stat.getVersion()));
}
- @NotNull
+ @NonNull
private CompletableFuture<Optional<LocatorEntry>> getSchemaLocator(String schema) {
return locatorEntryCache.getWithStats(schema)
.thenApply(o ->
@@ -574,7 +574,7 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
});
}
- @NotNull
+ @NonNull
private CompletableFuture<Long> addEntry(LedgerHandle ledgerHandle, SchemaStorageFormat.SchemaEntry entry) {
final CompletableFuture<Long> future = new CompletableFuture<>();
ledgerHandle.asyncAddEntry(entry.toByteArray(),
@@ -590,7 +590,7 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
return future;
}
- @NotNull
+ @NonNull
private CompletableFuture<LedgerHandle> createLedger(String schemaId) {
Map<String, byte[]> metadata = LedgerMetadataUtils.buildMetadataForSchema(schemaId);
final CompletableFuture<LedgerHandle> future = new CompletableFuture<>();
@@ -616,7 +616,7 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
return future;
}
- @NotNull
+ @NonNull
private CompletableFuture<LedgerHandle> openLedger(Long ledgerId) {
final CompletableFuture<LedgerHandle> future = new CompletableFuture<>();
bookKeeper.asyncOpenLedger(
@@ -635,7 +635,7 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
return future;
}
- @NotNull
+ @NonNull
private CompletableFuture<Void> closeLedger(LedgerHandle ledgerHandle) {
CompletableFuture<Void> future = new CompletableFuture<>();
ledgerHandle.asyncClose((rc, handle, ctx) -> {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorageFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorageFactory.java
index 2c84805584d..58154deda8c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorageFactory.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorageFactory.java
@@ -18,14 +18,14 @@
*/
package org.apache.pulsar.broker.service.schema;
-import javax.validation.constraints.NotNull;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.common.protocol.schema.SchemaStorage;
+import org.jspecify.annotations.NonNull;
@SuppressWarnings("unused")
public class BookkeeperSchemaStorageFactory implements SchemaStorageFactory {
@Override
- @NotNull
+ @NonNull
public SchemaStorage create(PulsarService pulsar) {
return new BookkeeperSchemaStorage(pulsar);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java
index c1a394dcfbb..dcf7edee1a2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java
@@ -39,7 +39,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
-import javax.validation.constraints.NotNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.avro.Schema;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
@@ -60,6 +59,7 @@ import org.apache.pulsar.common.protocol.schema.StoredSchema;
import org.apache.pulsar.common.schema.LongSchemaVersion;
import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.common.util.FutureUtil;
+import org.jspecify.annotations.NonNull;
@Slf4j
public class SchemaRegistryServiceImpl implements SchemaRegistryService {
@@ -87,7 +87,7 @@ public class SchemaRegistryServiceImpl implements SchemaRegistryService {
}
@Override
- @NotNull
+ @NonNull
public CompletableFuture<SchemaAndMetadata> getSchema(String schemaId) {
return getSchema(schemaId, SchemaVersion.Latest).thenApply((schema) -> {
if (schema != null && schema.schema.isDeleted()) {
@@ -99,7 +99,7 @@ public class SchemaRegistryServiceImpl implements SchemaRegistryService {
}
@Override
- @NotNull
+ @NonNull
public CompletableFuture<SchemaAndMetadata> getSchema(String schemaId, SchemaVersion version) {
long start = this.clock.millis();
@@ -186,7 +186,7 @@ public class SchemaRegistryServiceImpl implements SchemaRegistryService {
}
@Override
- @NotNull
+ @NonNull
public CompletableFuture<SchemaVersion> putSchemaIfAbsent(String schemaId, SchemaData schema,
SchemaCompatibilityStrategy strategy) {
MutableLong start = new MutableLong(0);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaStorageFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaStorageFactory.java
index ec0021c1ea8..c29b754abf8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaStorageFactory.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaStorageFactory.java
@@ -18,11 +18,11 @@
*/
package org.apache.pulsar.broker.service.schema;
-import javax.validation.constraints.NotNull;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.common.protocol.schema.SchemaStorage;
+import org.jspecify.annotations.NonNull;
public interface SchemaStorageFactory {
- @NotNull
+ @NonNull
SchemaStorage create(PulsarService pulsar) throws Exception;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
index f191fc3a384..5be4a170db0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
@@ -96,8 +96,8 @@ import org.apache.pulsar.common.policies.path.PolicyPath;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.coordination.LockManager;
-import org.checkerframework.checker.nullness.qual.NonNull;
-import org.checkerframework.checker.nullness.qual.Nullable;
+import org.jspecify.annotations.NonNull;
+import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
index 0256b7988c6..cc317fa99f6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
@@ -35,7 +35,6 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Predicate;
-import javax.annotation.Nullable;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.LedgerEntry;
@@ -53,6 +52,7 @@ import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.RawMessage;
import org.apache.pulsar.client.impl.RawMessageImpl;
import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicUtils.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicUtils.java
index a7a5fd4ef11..543c7907e95 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicUtils.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicUtils.java
@@ -24,7 +24,6 @@ import com.google.common.annotations.Beta;
import java.util.Collections;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
-import javax.annotation.Nullable;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -35,6 +34,7 @@ import org.apache.commons.collections4.CollectionUtils;
import org.apache.pulsar.broker.service.Consumer;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer;
import org.apache.pulsar.common.util.FutureUtil;
+import org.jspecify.annotations.Nullable;
public class CompactedTopicUtils {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactionServiceFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactionServiceFactory.java
index 7bb30372e45..e1a3945f50a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactionServiceFactory.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactionServiceFactory.java
@@ -19,10 +19,10 @@
package org.apache.pulsar.compaction;
import java.util.concurrent.CompletableFuture;
-import javax.annotation.Nonnull;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.common.classification.InterfaceAudience;
import org.apache.pulsar.common.classification.InterfaceStability;
+import org.jspecify.annotations.NonNull;
@InterfaceAudience.Public
@InterfaceStability.Evolving
@@ -35,7 +35,7 @@ public interface CompactionServiceFactory extends AutoCloseable {
* the pulsar service instance
* @return a future represents the initialization result
*/
- CompletableFuture<Void> initialize(@Nonnull PulsarService pulsarService);
+ CompletableFuture<Void> initialize(@NonNull PulsarService pulsarService);
/**
* Create a new topic compaction service for topic.
@@ -44,5 +44,5 @@ public interface CompactionServiceFactory extends AutoCloseable {
* the topic name
* @return a future represents the topic compaction service
*/
- CompletableFuture<TopicCompactionService> newTopicCompactionService(@Nonnull String topic);
+ CompletableFuture<TopicCompactionService> newTopicCompactionService(@NonNull String topic);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarCompactionServiceFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarCompactionServiceFactory.java
index 90132461b4c..ff0fa0bcd12 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarCompactionServiceFactory.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarCompactionServiceFactory.java
@@ -22,12 +22,12 @@ import com.google.common.annotations.VisibleForTesting;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
import lombok.AccessLevel;
import lombok.Getter;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.PulsarService;
+import org.jspecify.annotations.NonNull;
+import org.jspecify.annotations.Nullable;
public class PulsarCompactionServiceFactory implements CompactionServiceFactory {
@@ -60,14 +60,14 @@ public class PulsarCompactionServiceFactory implements CompactionServiceFactory
}
@Override
- public CompletableFuture<Void> initialize(@Nonnull PulsarService pulsarService) {
+ public CompletableFuture<Void> initialize(@NonNull PulsarService pulsarService) {
Objects.requireNonNull(pulsarService);
this.pulsarService = pulsarService;
return CompletableFuture.completedFuture(null);
}
@Override
- public CompletableFuture<TopicCompactionService> newTopicCompactionService(@Nonnull String topic) {
+ public CompletableFuture<TopicCompactionService> newTopicCompactionService(@NonNull String topic) {
Objects.requireNonNull(topic);
PulsarTopicCompactionService pulsarTopicCompactionService =
new PulsarTopicCompactionService(topic, pulsarService.getBookKeeperClient(), () -> {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarTopicCompactionService.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarTopicCompactionService.java
index 27efcf9524f..e123c11fd79 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarTopicCompactionService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarTopicCompactionService.java
@@ -30,13 +30,13 @@ import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.function.Predicate;
import java.util.function.Supplier;
-import javax.annotation.Nonnull;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.Position;
import org.apache.pulsar.common.api.proto.BrokerEntryMetadata;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.util.FutureUtil;
+import org.jspecify.annotations.NonNull;
public class PulsarTopicCompactionService implements TopicCompactionService {
@@ -66,7 +66,7 @@ public class PulsarTopicCompactionService implements TopicCompactionService {
}
@Override
- public CompletableFuture<List<Entry>> readCompactedEntries(@Nonnull Position startPosition,
+ public CompletableFuture<List<Entry>> readCompactedEntries(@NonNull Position startPosition,
int numberOfEntriesToRead) {
Objects.requireNonNull(startPosition);
checkArgument(numberOfEntriesToRead > 0);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TopicCompactionService.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TopicCompactionService.java
index fdd6bebbdec..11ce916f3ea 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TopicCompactionService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TopicCompactionService.java
@@ -20,11 +20,11 @@ package org.apache.pulsar.compaction;
import java.util.List;
import java.util.concurrent.CompletableFuture;
-import javax.annotation.Nonnull;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.Position;
import org.apache.pulsar.common.classification.InterfaceAudience;
import org.apache.pulsar.common.classification.InterfaceStability;
+import org.jspecify.annotations.NonNull;
@InterfaceAudience.Public
@InterfaceStability.Evolving
@@ -45,7 +45,7 @@ public interface TopicCompactionService extends AutoCloseable {
* @param numberOfEntriesToRead the maximum number of entries to read.
* @return a future that will be completed with the list of entries, this list can be null.
*/
- CompletableFuture<List<Entry>> readCompactedEntries(@Nonnull Position startPosition, int numberOfEntriesToRead);
+ CompletableFuture<List<Entry>> readCompactedEntries(@NonNull Position startPosition, int numberOfEntriesToRead);
/**
* Read the last compacted entry from the TopicCompactionService.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerTestZKBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerTestZKBaseTest.java
index a78254df4aa..96d88032302 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerTestZKBaseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerTestZKBaseTest.java
@@ -26,7 +26,7 @@ import org.apache.pulsar.metadata.TestZKServer;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
-import org.jetbrains.annotations.NotNull;
+import org.jspecify.annotations.NonNull;
/**
* Multiple brokers with a real test Zookeeper server (instead of the mock server)
@@ -71,7 +71,7 @@ public abstract class MultiBrokerTestZKBaseTest extends MultiBrokerBaseTest {
.configurationMetadataStore(createMetadataStore(MetadataStoreConfig.CONFIGURATION_METADATA_STORE));
}
- @NotNull
+ @NonNull
protected MetadataStoreExtended createMetadataStore(String metadataStoreName) {
try {
MetadataStoreExtended store =
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
index 3d83ac86d6c..c9eb8b6ef7c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
@@ -21,7 +21,6 @@ package org.apache.pulsar.broker.intercept;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNotNull;
-
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.nio.charset.StandardCharsets;
@@ -52,7 +51,7 @@ import org.apache.pulsar.common.intercept.BrokerEntryMetadataInterceptor;
import org.apache.pulsar.common.intercept.BrokerEntryMetadataUtils;
import org.apache.pulsar.common.intercept.ManagedLedgerPayloadProcessor;
import org.apache.pulsar.common.protocol.Commands;
-import org.checkerframework.checker.nullness.qual.Nullable;
+import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java
index be204ff3cdd..035941a93de 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java
@@ -77,7 +77,7 @@ import org.apache.zookeeper.MockZooKeeper;
import org.apache.zookeeper.MockZooKeeperSession;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
-import org.jetbrains.annotations.NotNull;
+import org.jspecify.annotations.NonNull;
import org.mockito.Mockito;
import org.mockito.internal.util.MockUtil;
@@ -998,7 +998,7 @@ public class PulsarTestContext implements AutoCloseable {
}
}
- @NotNull
+ @NonNull
private static ManagedLedgerStorage createManagedLedgerStorage(BookKeeper bookKeeperClient,
ManagedLedgerFactory managedLedgerFactory) {
BookkeeperManagedLedgerStorageClass managedLedgerStorageClass =
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/MultiBrokerMetadataConsistencyTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/MultiBrokerMetadataConsistencyTest.java
index c998a1ba57c..e58ff6b8b2a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/MultiBrokerMetadataConsistencyTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/MultiBrokerMetadataConsistencyTest.java
@@ -31,7 +31,7 @@ import org.apache.pulsar.metadata.TestZKServer;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
-import org.jetbrains.annotations.NotNull;
+import org.jspecify.annotations.NonNull;
import org.testng.annotations.Test;
@Slf4j
@@ -93,7 +93,7 @@ public class MultiBrokerMetadataConsistencyTest extends MultiBrokerBaseTest {
.configurationMetadataStore(configurationStore);
}
- @NotNull
+ @NonNull
protected MetadataStoreExtended createMetadataStore(String name) {
try {
return MetadataStoreExtended.create(testZKServer.getConnectionString(),
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicImplTest.java
index 9d89207d06f..bb6f09d222e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicImplTest.java
@@ -34,8 +34,8 @@ import java.util.function.Supplier;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.pulsar.common.api.proto.MessageIdData;
-import org.checkerframework.checker.nullness.qual.NonNull;
-import org.checkerframework.checker.nullness.qual.Nullable;
+import org.jspecify.annotations.NonNull;
+import org.jspecify.annotations.Nullable;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
diff --git a/pulsar-client/pom.xml b/pulsar-client/pom.xml
index 9163b66647e..e8af1709571 100644
--- a/pulsar-client/pom.xml
+++ b/pulsar-client/pom.xml
@@ -179,6 +179,11 @@
<artifactId>jcip-annotations</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.jspecify</groupId>
+ <artifactId>jspecify</artifactId>
+ </dependency>
+
<dependency>
<groupId>com.github.spotbugs</groupId>
<artifactId>spotbugs-annotations</artifactId>
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java
index 1674f7469af..c52a70c6cd7 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java
@@ -52,7 +52,7 @@ import org.asynchttpclient.DefaultAsyncHttpClientConfig;
import org.asynchttpclient.Request;
import org.asynchttpclient.Response;
import org.asynchttpclient.channel.DefaultKeepAliveStrategy;
-import org.checkerframework.checker.nullness.qual.Nullable;
+import org.jspecify.annotations.Nullable;
@Slf4j
public class ControlledClusterFailover implements ServiceUrlProvider {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
index 341272cd69b..c63c21d6d37 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -52,7 +52,6 @@ import java.util.function.Predicate;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
-import javax.annotation.Nullable;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.client.api.BatchReceivePolicy;
@@ -77,6 +76,7 @@ import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.util.CompletableFutureCancellationHandler;
import org.apache.pulsar.common.util.ExceptionHandler;
import org.apache.pulsar.common.util.FutureUtil;
+import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
index d30c3de0fd7..5f7957d7f1d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
@@ -39,7 +39,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Function;
-import javax.annotation.Nullable;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Triple;
@@ -53,6 +52,7 @@ import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.collections.BitSetRecyclable;
import org.apache.pulsar.common.util.collections.ConcurrentBitSetRecyclable;
+import org.jspecify.annotations.Nullable;
/**
* Group the acknowledgements for a certain time and then sends them out in a single protobuf command.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index 27e099171f8..f5722838d52 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -50,7 +50,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
-import javax.annotation.Nullable;
import lombok.Builder;
import lombok.Getter;
import org.apache.commons.lang3.tuple.Pair;
@@ -96,6 +95,7 @@ import org.apache.pulsar.common.util.Backoff;
import org.apache.pulsar.common.util.BackoffBuilder;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.netty.EventLoopUtil;
+import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/pulsar-common/pom.xml b/pulsar-common/pom.xml
index 567c1c01de2..5277f0feb6b 100644
--- a/pulsar-common/pom.xml
+++ b/pulsar-common/pom.xml
@@ -82,6 +82,11 @@
<artifactId>guava</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.jspecify</groupId>
+ <artifactId>jspecify</artifactId>
+ </dependency>
+
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-handler</artifactId>
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
index 454eee0f966..b9ee70dc825 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
@@ -39,8 +39,8 @@ import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import javax.annotation.Nonnull;
import javax.annotation.concurrent.ThreadSafe;
+import org.jspecify.annotations.NonNull;
/**
* This class is aimed at simplifying work with {@code CompletableFuture}.
@@ -284,7 +284,7 @@ public class FutureUtil {
* @throws RejectedExecutionException if this task cannot be accepted for execution
* @throws NullPointerException if one of params is null
*/
- public static <T> @Nonnull CompletableFuture<T> composeAsync(Supplier<CompletableFuture<T>> futureSupplier,
+ public static <T> @NonNull CompletableFuture<T> composeAsync(Supplier<CompletableFuture<T>> futureSupplier,
Executor executor) {
Objects.requireNonNull(futureSupplier);
Objects.requireNonNull(executor);
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java
index ea7fae2fdf0..946fcd56c0f 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java
@@ -22,9 +22,9 @@ import static java.util.Objects.requireNonNull;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.function.Predicate;
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
+import org.jspecify.annotations.NonNull;
+import org.jspecify.annotations.Nullable;
/**
* Static convenience URI checker.
@@ -39,8 +39,8 @@ public class URIPreconditions {
* @param predicate User defined rule
* @throws IllegalArgumentException Illegal URI or failed in the user's rules
*/
- public static void checkURI(@Nonnull String uri,
- @Nonnull Predicate<URI> predicate) throws IllegalArgumentException {
+ public static void checkURI(@NonNull String uri,
+ @NonNull Predicate<URI> predicate) throws IllegalArgumentException {
checkURI(uri, predicate, null);
}
@@ -52,7 +52,7 @@ public class URIPreconditions {
* @throws IllegalArgumentException Illegal URI or failed in the user's rules
*/
public static void checkURIIfPresent(@Nullable String uri,
- @Nonnull Predicate<URI> predicate) throws IllegalArgumentException {
+ @NonNull Predicate<URI> predicate) throws IllegalArgumentException {
checkURIIfPresent(uri, predicate, null);
}
@@ -65,7 +65,7 @@ public class URIPreconditions {
* @throws IllegalArgumentException Illegal URI or failed in the user's rules
*/
public static void checkURIIfPresent(@Nullable String uri,
- @Nonnull Predicate<URI> predicate,
+ @NonNull Predicate<URI> predicate,
@Nullable String errorMessage) throws IllegalArgumentException {
if (uri == null || uri.length() == 0) {
return;
@@ -81,8 +81,8 @@ public class URIPreconditions {
* @param errorMessage Error message
* @throws IllegalArgumentException Illegal URI or failed in the user's rules
*/
- public static void checkURI(@Nonnull String uri,
- @Nonnull Predicate<URI> predicate,
+ public static void checkURI(@NonNull String uri,
+ @NonNull Predicate<URI> predicate,
@Nullable String errorMessage) throws IllegalArgumentException {
requireNonNull(uri, "uri");
requireNonNull(predicate, "predicate");
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java
index 94bfad1fbd2..8ad75cd4e72 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java
@@ -33,7 +33,7 @@ import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.StampedLock;
import java.util.function.Consumer;
import java.util.function.Predicate;
-import javax.annotation.Nullable;
+import org.jspecify.annotations.Nullable;
/**
* This implements a {@link BlockingQueue} backed by an array with no fixed capacity.
diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java
index c83648132d4..c5ac63af24a 100644
--- a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java
+++ b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java
@@ -60,7 +60,7 @@ import org.apache.pulsar.io.core.SinkContext;
import org.apache.pulsar.io.core.Source;
import org.apache.pulsar.io.core.SourceContext;
import org.awaitility.Awaitility;
-import org.jetbrains.annotations.NotNull;
+import org.jspecify.annotations.NonNull;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
@@ -200,7 +200,7 @@ public class JavaInstanceRunnableTest {
verify(record, times(1)).ack();
}
- @NotNull
+ @NonNull
private JavaInstanceRunnable getJavaInstanceRunnable(boolean autoAck,
org.apache.pulsar.functions.proto.Function.ProcessingGuarantees processingGuarantees) throws Exception {
FunctionDetails functionDetails = FunctionDetails.newBuilder()
diff --git a/pulsar-functions/runtime-all/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceDepsTest.java b/pulsar-functions/runtime-all/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceDepsTest.java
index e29f19a8eee..7551270eac6 100644
--- a/pulsar-functions/runtime-all/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceDepsTest.java
+++ b/pulsar-functions/runtime-all/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceDepsTest.java
@@ -82,6 +82,7 @@ public class JavaInstanceDepsTest {
&& !name.startsWith("org/apache/commons/codec")
&& !name.startsWith("com/google")
&& !name.startsWith("org/checkerframework")
+ && !name.startsWith("org/jspecify/annotations")
&& !name.startsWith("javax/annotation")
&& !name.startsWith("org/apache/logging/slf4j")
&& !name.startsWith("org/apache/logging/log4j")) {
diff --git a/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXSink.java b/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXSink.java
index 07eb372833b..e19506cfd8b 100644
--- a/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXSink.java
+++ b/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXSink.java
@@ -54,7 +54,7 @@ import org.apache.pulsar.io.core.Sink;
import org.apache.pulsar.io.core.SinkContext;
import org.apache.pulsar.io.core.annotations.Connector;
import org.apache.pulsar.io.core.annotations.IOType;
-import org.jetbrains.annotations.NotNull;
+import org.jspecify.annotations.NonNull;
@Connector(
name = "adx",
@@ -177,7 +177,7 @@ public class ADXSink implements Sink<byte[]> {
}
}
- private boolean hasStreamingSucceeded(@NotNull IngestionStatus status) {
+ private boolean hasStreamingSucceeded(@NonNull IngestionStatus status) {
switch (status.status) {
case Succeeded:
case Queued:
@@ -230,7 +230,7 @@ public class ADXSink implements Sink<byte[]> {
}
}
- private @NotNull ADXPulsarEvent getADXPulsarEvent(@NotNull Record<byte[]> record) throws Exception {
+ private @NonNull ADXPulsarEvent getADXPulsarEvent(@NonNull Record<byte[]> record) throws Exception {
ADXPulsarEvent event = new ADXPulsarEvent();
record.getEventTime().ifPresent(time -> event.setEventTime(Instant.ofEpochMilli(time)));
record.getKey().ifPresent(event::setKey);
@@ -254,7 +254,7 @@ public class ADXSink implements Sink<byte[]> {
};
}
- private ConnectionStringBuilder getConnectionStringBuilder(@NotNull ADXSinkConfig adxConfig) {
+ private ConnectionStringBuilder getConnectionStringBuilder(@NonNull ADXSinkConfig adxConfig) {
if (adxConfig.getManagedIdentityId() != null) {
if ("system".equalsIgnoreCase(adxConfig.getManagedIdentityId())) {
diff --git a/pulsar-io/azure-data-explorer/src/test/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkConfigTest.java b/pulsar-io/azure-data-explorer/src/test/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkConfigTest.java
index c05e3b159d8..63aeb1091c3 100644
--- a/pulsar-io/azure-data-explorer/src/test/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkConfigTest.java
+++ b/pulsar-io/azure-data-explorer/src/test/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkConfigTest.java
@@ -18,7 +18,7 @@
*/
package org.apache.pulsar.io.azuredataexplorer;
-import org.jetbrains.annotations.NotNull;
+import org.jspecify.annotations.NonNull;
import org.testng.annotations.Test;
import java.io.File;
@@ -88,7 +88,7 @@ public class ADXSinkConfigTest {
assertEquals(config.getBatchTimeMs(), 10000);
}
- @NotNull
+ @NonNull
private static Map<String, Object> getConfig() {
Map<String, Object> map = new HashMap<>();
map.put("clusterUrl", "https://somecluster.eastus.kusto.windows.net");
@@ -106,7 +106,7 @@ public class ADXSinkConfigTest {
return map;
}
- private @NotNull File readSinkConfig(String name) {
+ private @NonNull File readSinkConfig(String name) {
ClassLoader classLoader = getClass().getClassLoader();
return new File(Objects.requireNonNull(classLoader.getResource(name)).getFile());
}
diff --git a/pulsar-io/core/pom.xml b/pulsar-io/core/pom.xml
index faa522f79d7..a6b58bbddeb 100644
--- a/pulsar-io/core/pom.xml
+++ b/pulsar-io/core/pom.xml
@@ -36,6 +36,10 @@
<artifactId>pulsar-functions-api</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.jspecify</groupId>
+ <artifactId>jspecify</artifactId>
+ </dependency>
</dependencies>
<build>
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithExtensibleLoadManagerTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithExtensibleLoadManagerTest.java
index 3567c8264f1..f0f8a3bbcae 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithExtensibleLoadManagerTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithExtensibleLoadManagerTest.java
@@ -66,7 +66,7 @@ import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.net.ServiceURI;
import org.apache.pulsar.metadata.impl.ZKMetadataStore;
-import org.jetbrains.annotations.NotNull;
+import org.jspecify.annotations.NonNull;
import org.mockito.Mockito;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -134,7 +134,7 @@ public class ProxyWithExtensibleLoadManagerTest extends MultiBrokerBaseTest {
}
}
- @NotNull
+ @NonNull
private InetSocketAddress getSourceBrokerInetAddress(TopicName topicName) throws PulsarAdminException {
var srcBrokerUrl = admin.lookups().lookupTopic(topicName.toString());
var serviceUri = ServiceURI.create(srcBrokerUrl);