[cleanup] Optional usage (apache#6613)

This PR aims to fix or simplify `Optional` type usage logic: - Null value used for optional type. - Optional.isPresent() can be replaced with functional style expression - Optional call chain that can be simplified.
vikingcodes · Mar 28, 2020 · acc161b · acc161b
1 parent c9a8d58
commit acc161b
Show file tree

Hide file tree

Showing 21 changed files with 72 additions and 141 deletions.
diff --git a/...er-common/src/main/java/org/apache/pulsar/broker/authentication/utils/AuthTokenUtils.java b/...er-common/src/main/java/org/apache/pulsar/broker/authentication/utils/AuthTokenUtils.java
@@ -96,9 +96,7 @@ public static String createToken(Key signingKey, String subject, Optional<Date>
                 .setSubject(subject)
                 .signWith(signingKey);
 
-        if (expiryTime.isPresent()) {
-            builder.setExpiration(expiryTime.get());
-        }
+        expiryTime.ifPresent(builder::setExpiration);
 
         return builder.compact();
     }

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java
@@ -897,9 +897,7 @@ public Map<String, FailureDomain> getFailureDomains(
                 try {
                     Optional<FailureDomain> domain = failureDomainCache()
                             .get(joinPath(failureDomainRootPath, domainName));
-                    if (domain.isPresent()) {
-                        domains.put(domainName, domain.get());
-                    }
+                    domain.ifPresent(failureDomain -> domains.put(domainName, failureDomain));
                 } catch (Exception e) {
                     log.warn("Failed to get domain {}", domainName, e);
                 }

diff --git a/...ker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java b/...ker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java
@@ -186,21 +186,18 @@ private Path getNicSpeedPath(String nic) {
     }
 
     private double getTotalNicLimitKbps(List<String> nics) {
-        if (overrideBrokerNicSpeedGbps.isPresent()) {
-            // Use the override value as configured. Return the total max speed across all available NICs, converted
-            // from Gbps into Kbps
-            return overrideBrokerNicSpeedGbps.get() * nics.size() * 1024 * 1024;
-        }
-
-        // Nic speed is in Mbits/s, return kbits/s
-        return nics.stream().mapToDouble(s -> {
-            try {
-                return Double.parseDouble(new String(Files.readAllBytes(getNicSpeedPath(s))));
-            } catch (IOException e) {
-                LOG.error("Failed to read speed for nic " + s, e);
-                return 0d;
-            }
-        }).sum() * 1024;
+        // Use the override value as configured. Return the total max speed across all available NICs, converted
+        // from Gbps into Kbps
+        return overrideBrokerNicSpeedGbps.map(aDouble -> aDouble * nics.size() * 1024 * 1024)
+                .orElseGet(() -> nics.stream().mapToDouble(s -> {
+                    // Nic speed is in Mbits/s, return kbits/s
+                    try {
+                        return Double.parseDouble(new String(Files.readAllBytes(getNicSpeedPath(s))));
+                    } catch (IOException e) {
+                        LOG.error("Failed to read speed for nic " + s, e);
+                        return 0d;
+                    }
+                }).sum() * 1024);
     }
 
     private Path getNicTxPath(String nic) {

diff --git a/...er/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java b/...er/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java
@@ -68,12 +68,8 @@ public LoadManagerReport generateLoadReport() {
     @Override
     public Optional<ResourceUnit> getLeastLoaded(final ServiceUnitId serviceUnit) {
         Optional<String> leastLoadedBroker = loadManager.selectBrokerForAssignment(serviceUnit);
-        if (leastLoadedBroker.isPresent()) {
-            return Optional.of(new SimpleResourceUnit(getBrokerWebServiceUrl(leastLoadedBroker.get()),
-                    new PulsarResourceDescription()));
-        } else {
-            return Optional.empty();
-        }
+        return leastLoadedBroker.map(s -> new SimpleResourceUnit(getBrokerWebServiceUrl(s),
+                new PulsarResourceDescription()));
     }
 
     private String getBrokerWebServiceUrl(String broker) {

diff --git a/...main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleResourceAllocationPolicies.java b/...main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleResourceAllocationPolicies.java
@@ -62,11 +62,7 @@ private Optional<NamespaceIsolationPolicies> getIsolationPolicies(String cluster
     public boolean areIsolationPoliciesPresent(NamespaceName namespace) {
         try {
             Optional<NamespaceIsolationPolicies> policies = getIsolationPolicies(pulsar.getConfiguration().getClusterName());
-            if (policies.isPresent()) {
-                return policies.get().getPolicyByNamespace(namespace) != null;
-            } else {
-                return false;
-            }
+            return policies.filter(isolationPolicies -> isolationPolicies.getPolicyByNamespace(namespace) != null).isPresent();
         } catch (Exception e) {
             LOG.warn("IsIsolationPoliciesPresent: Unable to get the namespaceIsolationPolicies", e);
             return false;
@@ -76,11 +72,8 @@ public boolean areIsolationPoliciesPresent(NamespaceName namespace) {
     private Optional<NamespaceIsolationPolicy> getNamespaceIsolationPolicy(NamespaceName namespace) {
         try {
             Optional<NamespaceIsolationPolicies> policies =getIsolationPolicies(pulsar.getConfiguration().getClusterName());
-            if (!policies.isPresent()) {
-                return Optional.empty();
-            }
+            return policies.map(isolationPolicies -> isolationPolicies.getPolicyByNamespace(namespace));
 
-            return Optional.ofNullable(policies.get().getPolicyByNamespace(namespace));
         } catch (Exception e) {
             LOG.warn("Unable to get the namespaceIsolationPolicies", e);
             return Optional.empty();
@@ -100,11 +93,8 @@ public boolean isSecondaryBroker(NamespaceName namespace, String broker) {
     public boolean isSharedBroker(String broker) {
         try {
             Optional<NamespaceIsolationPolicies> policies = getIsolationPolicies(pulsar.getConfiguration().getClusterName());
-            if (!policies.isPresent()) {
-                return true;
-            }
+            return policies.map(isolationPolicies -> isolationPolicies.isSharedBroker(broker)).orElse(true);
 
-            return policies.get().isSharedBroker(broker);
         } catch (Exception e) {
             LOG.warn("isPrimaryForAnyNamespace", e);
         }

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
@@ -1186,9 +1186,7 @@ private void refreshBrokerPublishRate() {
     public void forEachTopic(Consumer<Topic> consumer) {
         topics.forEach((n, t) -> {
             Optional<Topic> topic = extractTopic(t);
-            if (topic.isPresent()) {
-                consumer.accept(topic.get());
-            }
+            topic.ifPresent(consumer::accept);
         });
     }
 
@@ -1691,29 +1689,27 @@ private void updateDynamicServiceConfiguration() {
         } catch (Exception e) {
             log.warn("Failed to read zookeeper path [{}]:", BROKER_SERVICE_CONFIGURATION_PATH, e);
         }
-        if (configCache.isPresent()) {
-            configCache.get().forEach((key, value) -> {
-                // validate field
-                if (dynamicConfigurationMap.containsKey(key) && dynamicConfigurationMap.get(key).validator != null) {
-                    if (!dynamicConfigurationMap.get(key).validator.test(value)) {
-                        log.error("Failed to validate dynamic config {} with value {}", key, value);
-                        throw new IllegalArgumentException(
-                                String.format("Failed to validate dynamic-config %s/%s", key, value));
-                    }
+        configCache.ifPresent(stringStringMap -> stringStringMap.forEach((key, value) -> {
+            // validate field
+            if (dynamicConfigurationMap.containsKey(key) && dynamicConfigurationMap.get(key).validator != null) {
+                if (!dynamicConfigurationMap.get(key).validator.test(value)) {
+                    log.error("Failed to validate dynamic config {} with value {}", key, value);
+                    throw new IllegalArgumentException(
+                            String.format("Failed to validate dynamic-config %s/%s", key, value));
                 }
-                // update field value
-                try {
-                    Field field = ServiceConfiguration.class.getDeclaredField(key);
-                    if (field != null && field.isAnnotationPresent(FieldContext.class)) {
-                        field.setAccessible(true);
-                        field.set(pulsar().getConfiguration(), FieldParser.value(value, field));
-                        log.info("Successfully updated {}/{}", key, value);
-                    }
-                } catch (Exception e) {
-                    log.warn("Failed to update service configuration {}/{}, {}", key, value, e.getMessage());
+            }
+            // update field value
+            try {
+                Field field = ServiceConfiguration.class.getDeclaredField(key);
+                if (field != null && field.isAnnotationPresent(FieldContext.class)) {
+                    field.setAccessible(true);
+                    field.set(pulsar().getConfiguration(), FieldParser.value(value, field));
+                    log.info("Successfully updated {}/{}", key, value);
                 }
-            });
-        }
+            } catch (Exception e) {
+                log.warn("Failed to update service configuration {}/{}, {}", key, value, e.getMessage());
+            }
+        }));
         // register a listener: it updates field value and triggers appropriate registered field-listener only if
         // field's value has been changed so, registered doesn't have to update field value in ServiceConfiguration
         dynamicConfigurationCache.registerListener(new ZooKeeperCacheListener<Map<String, String>>() {

diff --git a/...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -376,13 +376,9 @@ public CompletableFuture<Void> close() {
             this.delayedDeliveryTracker = Optional.empty();
         }
 
-        if (delayedDeliveryTracker.isPresent()) {
-            delayedDeliveryTracker.get().close();
-        }
+        delayedDeliveryTracker.ifPresent(DelayedDeliveryTracker::close);
 
-        if (dispatchRateLimiter.isPresent()) {
-            dispatchRateLimiter.get().close();
-        }
+        dispatchRateLimiter.ifPresent(DispatchRateLimiter::close);
 
         return disconnectAllConsumers();
     }

diff --git a/...org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/...org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -253,9 +253,7 @@ public synchronized void internalReadEntriesComplete(final List<Entry> entries,
                                     topic.getDispatchRateLimiter().get().tryDispatchPermit(totalMessages, totalBytes);
                                 }
 
-                                if (dispatchRateLimiter.isPresent()) {
-                                    dispatchRateLimiter.get().tryDispatchPermit(totalMessages, totalBytes);
-                                }
+                                dispatchRateLimiter.ifPresent(rateLimiter -> rateLimiter.tryDispatchPermit(totalMessages, totalBytes));
                             }
 
                             // Schedule a new read batch operation only after the previous batch has been written to the
@@ -541,9 +539,7 @@ public void initializeDispatchRateLimiterIfNeeded(Optional<Policies> policies) {
     @Override
     public CompletableFuture<Void> close() {
         IS_CLOSED_UPDATER.set(this, TRUE);
-        if (dispatchRateLimiter.isPresent()) {
-            dispatchRateLimiter.get().close();
-        }
+        dispatchRateLimiter.ifPresent(DispatchRateLimiter::close);
         return disconnectAllConsumers();
     }
 

diff --git a/...roker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/...roker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
@@ -388,9 +388,7 @@ public void readEntriesComplete(List<Entry> entries, Object ctx) {
                     continue;
                 }
 
-                if (dispatchRateLimiter.isPresent()) {
-                    dispatchRateLimiter.get().tryDispatchPermit(1, entry.getLength());
-                }
+                dispatchRateLimiter.ifPresent(rateLimiter -> rateLimiter.tryDispatchPermit(1, entry.getLength()));
 
                 // Increment pending messages for messages produced locally
                 PENDING_MESSAGES_UPDATER.incrementAndGet(this);
@@ -756,9 +754,7 @@ public synchronized CompletableFuture<Void> disconnect(boolean failIfHasBacklog)
         final CompletableFuture<Void> future = new CompletableFuture<>();
 
         super.disconnect(failIfHasBacklog).thenRun(() -> {
-            if (dispatchRateLimiter.isPresent()) {
-                dispatchRateLimiter.get().close();
-            }
+            dispatchRateLimiter.ifPresent(DispatchRateLimiter::close);
             future.complete(null);
         }).exceptionally(ex -> {
             Throwable t = (ex instanceof CompletionException ? ex.getCause() : ex);

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
@@ -894,13 +894,9 @@ private CompletableFuture<Void> delete(boolean failIfHasSubscriptions,
                                 public void deleteLedgerComplete(Object ctx) {
                                     brokerService.removeTopicFromCache(topic);
 
-                                    if (dispatchRateLimiter.isPresent()) {
-                                        dispatchRateLimiter.get().close();
-                                    }
+                                    dispatchRateLimiter.ifPresent(DispatchRateLimiter::close);
 
-                                    if (subscribeRateLimiter.isPresent()) {
-                                        subscribeRateLimiter.get().close();
-                                    }
+                                    subscribeRateLimiter.ifPresent(SubscribeRateLimiter::close);
 
                                     log.info("[{}] Topic deleted", topic);
                                     deleteFuture.complete(null);
@@ -982,18 +978,11 @@ public void closeComplete(Object ctx) {
                     // Everything is now closed, remove the topic from map
                     brokerService.removeTopicFromCache(topic);
 
-                    ReplicatedSubscriptionsController ctrl = replicatedSubscriptionsController.orElse(null);
-                    if (ctrl != null) {
-                        ctrl.close();
-                    }
+                    replicatedSubscriptionsController.ifPresent(ReplicatedSubscriptionsController::close);
 
-                    if (dispatchRateLimiter.isPresent()) {
-                        dispatchRateLimiter.get().close();
-                    }
+                    dispatchRateLimiter.ifPresent(DispatchRateLimiter::close);
 
-                    if (subscribeRateLimiter.isPresent()) {
-                        subscribeRateLimiter.get().close();
-                    }
+                    subscribeRateLimiter.ifPresent(SubscribeRateLimiter::close);
 
                     log.info("[{}] Topic closed", topic);
                     closeFuture.complete(null);

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
@@ -81,7 +81,7 @@ public static <T> MessageImpl<T> create(MessageMetadata.Builder msgMetadataBuild
     // Constructor for incoming message
     MessageImpl(String topic, MessageIdImpl messageId, MessageMetadata msgMetadata,
                 ByteBuf payload, ClientCnx cnx, Schema<T> schema) {
-        this(topic, messageId, msgMetadata, payload, null, cnx, schema);
+        this(topic, messageId, msgMetadata, payload, Optional.empty(), cnx, schema);
     }
 
     MessageImpl(String topic, MessageIdImpl messageId, MessageMetadata msgMetadata, ByteBuf payload,

diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
@@ -1108,9 +1108,7 @@ public static ByteBuf newGetSchema(long requestId, String topic, Optional<Schema
         CommandGetSchema.Builder schema = CommandGetSchema.newBuilder()
             .setRequestId(requestId);
         schema.setTopic(topic);
-        if (version.isPresent()) {
-            schema.setSchemaVersion(ByteString.copyFrom(version.get().bytes()));
-        }
+        version.ifPresent(schemaVersion -> schema.setSchemaVersion(ByteString.copyFrom(schemaVersion.bytes())));
 
         CommandGetSchema getSchema = schema.build();
 

diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Markers.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Markers.java
@@ -49,9 +49,7 @@ private static ByteBuf newMessage(MarkerType markerType, Optional<String> restri
         msgMetadataBuilder.setSequenceId(0);
         msgMetadataBuilder.setMarkerType(markerType.getNumber());
 
-        if (restrictToCluster.isPresent()) {
-            msgMetadataBuilder.addReplicateTo(restrictToCluster.get());
-        }
+        restrictToCluster.ifPresent(msgMetadataBuilder::addReplicateTo);
 
         MessageMetadata msgMetadata = msgMetadataBuilder.build();
         try {

diff --git a/...overy-service/src/main/java/org/apache/pulsar/discovery/service/server/ServiceConfig.java b/...overy-service/src/main/java/org/apache/pulsar/discovery/service/server/ServiceConfig.java
@@ -168,7 +168,7 @@ public void setWebServicePortTls(Optional<Integer> webServicePortTls) {
 
     @Deprecated
     public boolean isTlsEnabled() {
-        return tlsEnabled || webServicePortTls != null || servicePortTls != null;
+        return tlsEnabled || webServicePortTls.isPresent() || servicePortTls.isPresent();
     }
 
     @Deprecated

diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
@@ -294,9 +294,7 @@ public void write(Record<T> record) {
         } else {
             // It is coming from some source
             Optional<Long> eventTime = sinkRecord.getSourceRecord().getEventTime();
-            if (eventTime.isPresent()) {
-                msg.eventTime(eventTime.get());
-            }
+            eventTime.ifPresent(msg::eventTime);
         }
 
         pulsarSinkProcessor.sendOutputMessage(msg, record);

diff --git a/...les/src/main/java/org/apache/pulsar/functions/api/examples/ConfigBasedAppendFunction.java b/...les/src/main/java/org/apache/pulsar/functions/api/examples/ConfigBasedAppendFunction.java
@@ -32,10 +32,6 @@ public String process(String input, Context context) {
         String key = "config-key";
         Optional<Object> appendValue = context.getUserConfigValue(key);
 
-        if (appendValue.isPresent()) {
-            return input + appendValue.get();
-        } else {
-            return input + "!";
-        }
+        return appendValue.map(o -> input + (String) o).orElseGet(() -> input + "!");
     }
 }
diff --git a/...r-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/FunctionAuthUtils.java b/...r-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/FunctionAuthUtils.java
@@ -25,9 +25,8 @@
 public final class FunctionAuthUtils {
 
     public static final FunctionAuthData getFunctionAuthData(Optional<Function.FunctionAuthenticationSpec> functionAuthenticationSpec) {
-        if (functionAuthenticationSpec.isPresent()) {
-            return FunctionAuthData.builder().data(functionAuthenticationSpec.get().getData().toByteArray()).build();
-        }
-        return null;
+        return functionAuthenticationSpec
+                .map(authenticationSpec -> FunctionAuthData.builder().data(authenticationSpec.getData().toByteArray()).build())
+                .orElse(null);
     }
 }
diff --git a/...me/src/main/java/org/apache/pulsar/functions/auth/KubernetesSecretsTokenAuthProvider.java b/...me/src/main/java/org/apache/pulsar/functions/auth/KubernetesSecretsTokenAuthProvider.java
@@ -254,11 +254,7 @@ public Optional<FunctionAuthData> updateAuthData(Function.FunctionDetails funcDe
                                                      AuthenticationDataSource authenticationDataSource) throws Exception {
 
         String secretId;
-        if (existingFunctionAuthData.isPresent()) {
-            secretId = new String(existingFunctionAuthData.get().getData());
-        } else {
-            secretId = RandomStringUtils.random(5, true, true).toLowerCase();
-        }
+        secretId = existingFunctionAuthData.map(functionAuthData -> new String(functionAuthData.getData())).orElseGet(() -> RandomStringUtils.random(5, true, true).toLowerCase());
 
         String token;
         try {

diff --git a/...tions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java b/...tions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
@@ -198,12 +198,10 @@ public void registerFunction(final String tenant,
                             Optional<FunctionAuthData> functionAuthData = functionAuthProvider
                                     .cacheAuthData(finalFunctionDetails, clientAuthenticationDataHttps);
 
-                            if (functionAuthData.isPresent()) {
-                                functionMetaDataBuilder.setFunctionAuthSpec(
-                                        Function.FunctionAuthenticationSpec.newBuilder()
-                                                .setData(ByteString.copyFrom(functionAuthData.get().getData()))
-                                                .build());
-                            }
+                            functionAuthData.ifPresent(authData -> functionMetaDataBuilder.setFunctionAuthSpec(
+                                    Function.FunctionAuthenticationSpec.newBuilder()
+                                            .setData(ByteString.copyFrom(authData.getData()))
+                                            .build()));
                         } catch (Exception e) {
                             log.error("Error caching authentication data for {} {}/{}/{}",
                                     ComponentTypeUtils.toString(componentType), tenant, namespace, functionName, e);