[Issue 7933][pulsar-broker-common] Add some logging to improve Authen…

…tication debugging and Fix typos in code "occured" -> "occurred" (apache#7934)

Fixes apache#7933 

### Motivation
Newbie trying to make his first contribution to this projection :-)

### Modifications

Added some logging to Authentication Service to help debugging when there are more than one AuthN provider.
When I did that I noticed a typo ('occured' should have two r's) so I decided to try to fix all of them

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationService.java

@@ -90,6 +90,9 @@ public String authenticateHttpRequest(HttpServletRequest request) throws Authent
             try {
                 return provider.authenticate(authData);
             } catch (AuthenticationException e) {
+                if (LOG.isDebugEnabled()) {
+                    LOG.debug("Authentication failed for provider " + provider.getAuthMethodName() + ": " + e.getMessage(), e);
+                }
                 // Store the exception so we can throw it later instead of a generic one
                 authenticationException = e;
             }

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java

@@ -300,7 +300,7 @@ public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String rol
             } else {
                 if (log.isDebugEnabled()) {
                     log.debug(
-                            "Topic [{}] Role [{}] exception occured while trying to check Produce permissions. {}",
+                            "Topic [{}] Role [{}] exception occurred while trying to check Produce permissions. {}",
                             topicName.toString(), role, ex.getMessage());
                 }
             }
@@ -313,7 +313,7 @@ public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String rol
                 } else {
                     if (log.isDebugEnabled()) {
                         log.debug(
-                                "Topic [{}] Role [{}] exception occured while trying to check Consume permissions. {}",
+                                "Topic [{}] Role [{}] exception occurred while trying to check Consume permissions. {}",
                                 topicName.toString(), role, e.getMessage());
 
                     }

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java

@@ -191,7 +191,7 @@ public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String rol
             } else {
                 if (log.isDebugEnabled()) {
                     log.debug(
-                            "Topic [{}] Role [{}] exception occured while trying to check Produce permissions. {}",
+                            "Topic [{}] Role [{}] exception occurred while trying to check Produce permissions. {}",
                             topicName.toString(), role, ex.getMessage());
                 }
             }
@@ -204,7 +204,7 @@ public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String rol
                 } else {
                     if (log.isDebugEnabled()) {
                         log.debug(
-                                "Topic [{}] Role [{}] exception occured while trying to check Consume permissions. {}",
+                                "Topic [{}] Role [{}] exception occurred while trying to check Consume permissions. {}",
                                 topicName.toString(), role, e.getMessage());
 
                     }

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

@@ -168,7 +168,7 @@ protected synchronized CompletableFuture<Void> closeProducerAsync() {
         }).exceptionally(ex -> {
             long waitTimeMs = backOff.next();
             log.warn(
-                    "[{}][{} -> {}] Exception: '{}' occured while trying to close the producer. retrying again in {} s",
+                    "[{}][{} -> {}] Exception: '{}' occurred while trying to close the producer. retrying again in {} s",
                     topicName, localCluster, remoteCluster, ex.getMessage(), waitTimeMs / 1000.0);
             // BackOff before retrying
             brokerService.executor().schedule(this::closeProducerAsync, waitTimeMs, TimeUnit.MILLISECONDS);

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

@@ -383,7 +383,7 @@ protected void handleLookup(CommandLookupTopic lookup) {
                 }
                 return null;
             }).exceptionally(ex -> {
-                final String msg = "Exception occured while trying to authorize lookup";
+                final String msg = "Exception occurred while trying to authorize lookup";
                 log.warn("[{}] {} with role {} on topic {}", remoteAddress, msg, getPrincipal(), topicName, ex);
                 ctx.writeAndFlush(newLookupErrorResponse(ServerError.AuthorizationError, msg, requestId));
                 lookupSemaphore.release();
@@ -458,7 +458,7 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa
                 }
                 return null;
             }).exceptionally(ex -> {
-                final String msg = "Exception occured while trying to authorize get Partition Metadata";
+                final String msg = "Exception occurred while trying to authorize get Partition Metadata";
                 log.warn("[{}] {} with role {} on topic {}", remoteAddress, msg, getPrincipal(), topicName);
                 ctx.writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg, requestId));
                 lookupSemaphore.release();

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

@@ -138,7 +138,7 @@ public void testOwnershipAfterSetup() {
                 assertTrue(pulsarServices[0].getNamespaceService().registerSLANamespace());
             } catch (PulsarServerException e) {
                 e.printStackTrace();
-                log.error("Exception occured", e);
+                log.error("Exception occurred", e);
                 fail("SLA Namespace should have been owned by the broker, Exception.", e);
             }
         }

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java

@@ -1348,7 +1348,7 @@ private List<Message<byte[]>> getIndividualMsgsFromBatch(String topic, String ms
                 ret.add(new MessageImpl<>(topic, batchMsgId, properties, singleMessagePayload,
                         Schema.BYTES, msgMetadataBuilder));
             } catch (Exception ex) {
-                log.error("Exception occured while trying to get BatchMsgId: {}", batchMsgId, ex);
+                log.error("Exception occurred while trying to get BatchMsgId: {}", batchMsgId, ex);
             }
             singleMessageMetadataBuilder.recycle();
         }

pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailerTest.java

@@ -151,7 +151,7 @@ public Boolean answer(InvocationOnMock invocationOnMock) throws Throwable {
 
         functionAssignmentTailer.start();
 
-        // verify no errors occured
+        // verify no errors occurred
         verify(errorNotifier, times(0)).triggerError(any());
 
         messageList.add(message1);

pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionRuntimeManagerTest.java

@@ -622,7 +622,7 @@ public Boolean answer(InvocationOnMock invocationOnMock) throws Throwable {
                         Function.Instance.newBuilder().setFunctionMetaData(function1).setInstanceId(0)
                                 .build()));
 
-        // verify no errors occured
+        // verify no errors occurred
         verify(errorNotifier, times(0)).triggerError(any());
     }
 

pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRolesEnforcementTest.java

@@ -196,13 +196,13 @@ public void testIncorrectRoles() throws Exception {
 
         // Step 2: Try to use proxy Client as a normal Client - expect exception
         PulsarClient proxyClient = createPulsarClient(pulsar.getBrokerServiceUrl(), proxyAuthParams);
-        boolean exceptionOccured = false;
+        boolean exceptionOccurred = false;
         try {
             proxyClient.newConsumer().topic(topicName).subscriptionName(subscriptionName).subscribe();
         } catch (Exception ex) {
-            exceptionOccured = true;
+            exceptionOccurred = true;
         }
-        Assert.assertTrue(exceptionOccured);
+        Assert.assertTrue(exceptionOccurred);
 
         // Step 3: Run Pulsar Proxy and pass proxy params as client params - expect exception
         ProxyConfiguration proxyConfig = new ProxyConfiguration();
@@ -224,14 +224,14 @@ public void testIncorrectRoles() throws Exception {
         proxyService.start();
 
         proxyClient = createPulsarClient(proxyService.getServiceUrl(), proxyAuthParams);
-        exceptionOccured = false;
+        exceptionOccurred = false;
         try {
             proxyClient.newConsumer().topic(topicName).subscriptionName(subscriptionName).subscribe();
         } catch (Exception ex) {
-            exceptionOccured = true;
+            exceptionOccurred = true;
         }
 
-        Assert.assertTrue(exceptionOccured);
+        Assert.assertTrue(exceptionOccurred);
 
         // Step 4: Pass correct client params
         proxyClient = createPulsarClient(proxyService.getServiceUrl(), clientAuthParams);