BP-69: Convert stream/distributedlog modules to slog (phase 5)

bookkeeper-dist/src/main/resources/LICENSE-all.bin.txt

@@ -216,6 +216,7 @@ Apache Software License, Version 2.
 - lib/commons-codec-commons-codec-1.18.0.jar [6]
 - lib/commons-io-commons-io-2.19.0.jar [8]
 - lib/commons-logging-commons-logging-1.3.5.jar [10]
+- lib/io.github.merlimat.slog-slog-0.9.8.jar [64]
 - lib/io.netty-netty-buffer-4.2.12.Final.jar [11]
 - lib/io.netty-netty-codec-base-4.2.12.Final.jar [11]
 - lib/io.netty-netty-codec-compression-4.2.12.Final.jar [11]
@@ -422,6 +423,7 @@ Apache Software License, Version 2.
 [61] Source available at https://github.com/apache/commons-text/tree/rel/commons-text-1.13.1
 [62] Source available at https://github.com/apache/commons-beanutils/tree/rel/commons-beanutils-1.11.0
 [63] Source available at https://github.com/googleapis/sdk-platform-java/tree/v2.53.0/api-common-java
+[64] Source available at https://github.com/merlimat/slog/tree/v0.9.8
 ------------------------------------------------------------------------------------
 lib/io.netty-netty-codec-base-4.2.12.Final.jar bundles some 3rd party dependencies
 

bookkeeper-dist/src/main/resources/LICENSE-bkctl.bin.txt

@@ -216,6 +216,7 @@ Apache Software License, Version 2.
 - lib/commons-codec-commons-codec-1.18.0.jar [6]
 - lib/commons-io-commons-io-2.19.0.jar [8]
 - lib/commons-logging-commons-logging-1.3.5.jar [10]
+- lib/io.github.merlimat.slog-slog-0.9.8.jar [59]
 - lib/io.netty-netty-buffer-4.2.12.Final.jar [11]
 - lib/io.netty-netty-codec-base-4.2.12.Final.jar [11]
 - lib/io.netty-netty-common-4.2.12.Final.jar [11]
@@ -355,6 +356,7 @@ Apache Software License, Version 2.
 [56] Source available at https://github.com/apache/commons-text/tree/rel/commons-text-1.13.1
 [57] Source available at https://github.com/apache/commons-beanutils/tree/rel/commons-beanutils-1.11.0
 [58] Source available at https://github.com/googleapis/sdk-platform-java/tree/v2.53.0/api-common-java
+[59] Source available at https://github.com/merlimat/slog/tree/v0.9.8
 ------------------------------------------------------------------------------------
 lib/io.netty-netty-codec-base-4.2.12.Final.jar bundles some 3rd party dependencies
 

bookkeeper-dist/src/main/resources/LICENSE-server.bin.txt

@@ -216,6 +216,7 @@ Apache Software License, Version 2.
 - lib/commons-codec-commons-codec-1.18.0.jar [6]
 - lib/commons-io-commons-io-2.19.0.jar [8]
 - lib/commons-logging-commons-logging-1.3.5.jar [10]
+- lib/io.github.merlimat.slog-slog-0.9.8.jar [63]
 - lib/io.netty-netty-buffer-4.2.12.Final.jar [11]
 - lib/io.netty-netty-codec-base-4.2.12.Final.jar [11]
 - lib/io.netty-netty-codec-compression-4.2.12.Final.jar [11]
@@ -417,6 +418,7 @@ Apache Software License, Version 2.
 [60] Source available at https://github.com/apache/commons-text/tree/rel/commons-text-1.13.1
 [61] Source available at https://github.com/apache/commons-beanutils/tree/rel/commons-beanutils-1.11.0
 [62] Source available at https://github.com/googleapis/sdk-platform-java/tree/v2.53.0/api-common-java
+[63] Source available at https://github.com/merlimat/slog/tree/v0.9.8
 ------------------------------------------------------------------------------------
 lib/io.netty-netty-codec-base-4.2.12.Final.jar bundles some 3rd party dependencies
 

lombok.config

@@ -0,0 +1,24 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+# this is the top level Lombok configuration file
+# see https://projectlombok.org/features/configuration for reference
+
+config.stopBubbling = true
+lombok.log.custom.declaration = io.github.merlimat.slog.Logger io.github.merlimat.slog.Logger.get(TYPE)

pom.xml

@@ -192,6 +192,7 @@
     <rocksdb.version>9.9.3</rocksdb.version>
     <shrinkwrap.version>3.3.0</shrinkwrap.version>
     <slf4j.version>2.0.12</slf4j.version>
+    <slog.version>0.9.8</slog.version>
     <snakeyaml.version>2.0</snakeyaml.version>
     <spotbugs-annotations.version>4.6.0</spotbugs-annotations.version>
     <spotless.version>2.43.0</spotless.version>
@@ -273,6 +274,11 @@
       </dependency>
 
       <!-- logging dependencies -->
+      <dependency>
+        <groupId>io.github.merlimat.slog</groupId>
+        <artifactId>slog</artifactId>
+        <version>${slog.version}</version>
+      </dependency>
       <dependency>
         <groupId>org.slf4j</groupId>
         <artifactId>slf4j-bom</artifactId>
@@ -882,6 +888,10 @@
     </dependency>
 
     <!-- compilation dependencies (available at all classpaths) -->
+    <dependency>
+      <groupId>io.github.merlimat.slog</groupId>
+      <artifactId>slog</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>

stream/bk-grpc-name-resolver/src/main/java/org/apache/bookkeeper/grpc/resolver/BKRegistrationNameResolverProvider.java

@@ -24,7 +24,7 @@
 import io.grpc.NameResolverProvider;
 import java.net.URI;
 import javax.annotation.Nullable;
-import lombok.extern.slf4j.Slf4j;
+import lombok.CustomLog;
 import org.apache.bookkeeper.common.net.ServiceURI;
 import org.apache.bookkeeper.common.resolver.NameResolverFactoryProvider;
 import org.apache.bookkeeper.common.resolver.NameResolverProviderFactory;
@@ -42,7 +42,7 @@
 // https://github.com/grpc/grpc-java/issues/7133#issuecomment-680981331
 // Skipping the migration for now, will have to deal with this later when GRPC team
 // finalizes their API.
-@Slf4j
+@CustomLog
 public class BKRegistrationNameResolverProvider extends NameResolverFactoryProvider {
 
     @Override
@@ -63,8 +63,10 @@ public NameResolver newNameResolver(URI targetUri, NameResolver.Args args) {
             serviceURI = ServiceURI.create(targetUri);
         } catch (NullPointerException | IllegalArgumentException e) {
             // invalid uri here, so return null to allow grpc to use other name resolvers
-            log.info("BKRegistrationNameResolverProvider doesn't know how to resolve {} : cause {}",
-                targetUri, e.getMessage());
+            log.info()
+                .attr("targetUri", targetUri)
+                .exceptionMessage(e)
+                .log("BKRegistrationNameResolverProvider doesn't know how to resolve target URI");
             return null;
         }
 
@@ -73,7 +75,10 @@ public NameResolver newNameResolver(URI targetUri, NameResolver.Args args) {
             clientDriver = MetadataDrivers.getClientDriver(serviceURI.getUri());
             return new BKRegistrationNameResolver(clientDriver, serviceURI.getUri());
         } catch (IllegalArgumentException iae) {
-            log.error("Unknown service uri : {}", serviceURI, iae);
+            log.error()
+                .attr("serviceURI", serviceURI)
+                .exception(iae)
+                .log("Unknown service uri");
             return null;
         }
     }

stream/clients/java/all/src/main/java/org/apache/bookkeeper/clients/SimpleStorageClientImpl.java

@@ -28,7 +28,7 @@
 import io.netty.buffer.ByteBuf;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
-import lombok.extern.slf4j.Slf4j;
+import lombok.CustomLog;
 import org.apache.bookkeeper.api.StorageClient;
 import org.apache.bookkeeper.api.exceptions.ApiException;
 import org.apache.bookkeeper.api.kv.PTable;
@@ -50,7 +50,7 @@
 /**
  * The implementation of {@link StorageClient} client.
  */
-@Slf4j
+@CustomLog
 public class SimpleStorageClientImpl extends SimpleClientBase implements StorageClient {
 
     private static final String COMPONENT_NAME = SimpleStorageClientImpl.class.getSimpleName();
@@ -117,8 +117,11 @@ private void openTableImpl(String namespaceName,
         ).thenCompose(resp -> {
             if (StatusCode.SUCCESS == resp.getCode()) {
                 StreamProperties streamProps = resp.getStreamProps();
-                log.info("Retrieved table properties for table {}/{} : {}",
-                    namespaceName, streamName, streamProps);
+                log.info()
+                    .attr("namespace", namespaceName)
+                    .attr("streamName", streamName)
+                    .attr("streamProps", streamProps)
+                    .log("Retrieved table properties for table");
                 if (StorageType.TABLE != streamProps.getStreamConf().getStorageType()) {
                     return FutureUtils.exception(new ApiException(
                         "Can't open a non-table storage entity : " + streamProps.getStreamConf().getStorageType()));

stream/clients/java/all/src/main/java/org/apache/bookkeeper/clients/StorageClientImpl.java

@@ -22,7 +22,7 @@
 import io.netty.buffer.ByteBuf;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
-import lombok.extern.slf4j.Slf4j;
+import lombok.CustomLog;
 import org.apache.bookkeeper.api.StorageClient;
 import org.apache.bookkeeper.api.exceptions.ApiException;
 import org.apache.bookkeeper.api.kv.PTable;
@@ -44,7 +44,7 @@
 /**
  * The implementation of {@link StorageClient} client.
  */
-@Slf4j
+@CustomLog
 public class StorageClientImpl extends AbstractAutoAsyncCloseable implements StorageClient {
 
     private static final String COMPONENT_NAME = StorageClientImpl.class.getSimpleName();
@@ -122,9 +122,11 @@ private void openTableImpl(String namespaceName,
                                CompletableFuture<PTable<ByteBuf, ByteBuf>> future) {
         FutureUtils.proxyTo(
             getStreamProperties(namespaceName, tableName).thenComposeAsync(props -> {
-                if (log.isInfoEnabled()) {
-                    log.info("Retrieved table properties for table {}/{} : {}", namespaceName, tableName, props);
-                }
+                log.info()
+                    .attr("namespace", namespaceName)
+                    .attr("tableName", tableName)
+                    .attr("props", props)
+                    .log("Retrieved table properties for table");
                 if (StorageType.TABLE != props.getStreamConf().getStorageType()) {
                     return FutureUtils.exception(new ApiException(
                         "Can't open a non-table storage entity : " + props.getStreamConf().getStorageType())
@@ -167,7 +169,7 @@ public void close() {
         try {
             super.close(1, TimeUnit.SECONDS);
         } catch (Exception e) {
-            log.warn("Encountered exceptions on closing the storage client", e);
+            log.warn().exception(e).log("Encountered exceptions on closing the storage client");
         }
         scheduler.forceShutdown(100, TimeUnit.MILLISECONDS);
     }

stream/clients/java/all/src/main/java/org/apache/bookkeeper/clients/admin/StorageAdminClientImpl.java

@@ -21,7 +21,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Supplier;
-import lombok.extern.slf4j.Slf4j;
+import lombok.CustomLog;
 import org.apache.bookkeeper.api.StorageClient;
 import org.apache.bookkeeper.clients.StorageClientImpl;
 import org.apache.bookkeeper.clients.config.StorageClientSettings;
@@ -38,7 +38,7 @@
 /**
  * A storage admin client.
  */
-@Slf4j
+@CustomLog
 public class StorageAdminClientImpl extends AbstractAutoAsyncCloseable implements StorageAdminClient {
 
     // clients
@@ -127,13 +127,9 @@ public void close() {
         try {
             closeAsync().get();
         } catch (InterruptedException e) {
-            if (log.isDebugEnabled()) {
-                log.debug("Interrupted on closing stream admin client", e);
-            }
+            log.debug().exception(e).log("Interrupted on closing stream admin client");
         } catch (ExecutionException e) {
-            if (log.isDebugEnabled()) {
-                log.debug("Failed to cloe stream admin client", e);
-            }
+            log.debug().exception(e).log("Failed to cloe stream admin client");
         }
     }
 }

stream/clients/java/all/src/test/java/org/apache/bookkeeper/clients/StorageClientImplTest.java

@@ -31,7 +31,7 @@
 import static org.mockito.Mockito.when;
 
 import io.netty.buffer.ByteBuf;
-import lombok.extern.slf4j.Slf4j;
+import lombok.CustomLog;
 import org.apache.bookkeeper.api.exceptions.ApiException;
 import org.apache.bookkeeper.api.kv.PTable;
 import org.apache.bookkeeper.api.kv.Table;
@@ -51,7 +51,7 @@
  * Unit test {@link StorageClientImpl}.
  */
 @RunWith(MockitoJUnitRunner.class)
-@Slf4j
+@CustomLog
 public class StorageClientImplTest extends GrpcClientTestBase {
 
     private static final String NAMESPACE = "test-namespace";

stream/clients/java/base/src/main/java/org/apache/bookkeeper/clients/impl/channel/StorageServerChannelManager.java

@@ -24,14 +24,14 @@
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Function;
 import javax.annotation.Nullable;
-import lombok.extern.slf4j.Slf4j;
+import lombok.CustomLog;
 import org.apache.bookkeeper.clients.config.StorageClientSettings;
 import org.apache.bookkeeper.stream.proto.common.Endpoint;
 
 /**
  * A manager manages channels to range servers.
  */
-@Slf4j
+@CustomLog
 public class StorageServerChannelManager implements AutoCloseable {
 
     private final ReentrantReadWriteLock lock;
@@ -69,21 +69,21 @@ public boolean addStorageServer(Endpoint endpoint, StorageServerChannel channel)
         lock.readLock().lock();
         try {
             if (closed) {
-                log.warn("Skip adding channel {} of range server {} since the channel manager is already closed",
-                    channel, endpoint);
+                log.warn()
+                    .attr("channel", channel)
+                    .attr("endpoint", endpoint)
+                    .log("Skip adding channel since the channel manager is already closed");
                 channel.close();
                 return false;
             }
 
             StorageServerChannel oldChannel = channels.putIfAbsent(endpoint, channel);
             if (null != oldChannel) {
-                if (log.isDebugEnabled()) {
-                    log.debug("KeyRange server ({}) already existed in the channel manager.", endpoint);
-                }
+                log.debug().attr("endpoint", endpoint).log("KeyRange server already existed in the channel manager.");
                 channel.close();
                 return false;
             } else {
-                log.info("Added range server ({}) into the channel manager.", endpoint);
+                log.info().attr("endpoint", endpoint).log("Added range server into the channel manager.");
                 return true;
             }
         } finally {
@@ -117,8 +117,10 @@ public StorageServerChannel removeChannel(Endpoint endpoint, StorageServerChanne
         lock.readLock().lock();
         try {
             if (closed) {
-                log.warn("Skip removing channel {} of range server {} since the channel manager is already closed",
-                    channel, endpoint);
+                log.warn()
+                    .attr("channel", channel)
+                    .attr("endpoint", endpoint)
+                    .log("Skip removing channel since the channel manager is already closed");
                 return null;
             }
 
@@ -133,12 +135,12 @@ public StorageServerChannel removeChannel(Endpoint endpoint, StorageServerChanne
                 }
             }
             if (null == channelRemoved) {
-                if (log.isDebugEnabled()) {
-                    log.debug("No channel associated with endpoint {} to be removed.", endpoint);
-                }
+                log.debug().attr("endpoint", endpoint).log("No channel associated with endpoint to be removed.");
             } else {
-                log.info("Removed channel {} for range server {} successfully",
-                    channelRemoved, endpoint);
+                log.info()
+                    .attr("channel", channelRemoved)
+                    .attr("endpoint", endpoint)
+                    .log("Removed channel for range server successfully");
             }
             if (null != channelRemoved) {
                 channelRemoved.close();

stream/clients/java/base/src/main/java/org/apache/bookkeeper/clients/impl/container/StorageContainerChannel.java

@@ -25,7 +25,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.concurrent.GuardedBy;
-import lombok.extern.slf4j.Slf4j;
+import lombok.CustomLog;
 import org.apache.bookkeeper.clients.exceptions.StorageContainerException;
 import org.apache.bookkeeper.clients.impl.channel.StorageServerChannel;
 import org.apache.bookkeeper.clients.impl.channel.StorageServerChannelManager;
@@ -43,7 +43,7 @@
 /**
  * A client place holder for managing information of storage containers.
  */
-@Slf4j
+@CustomLog
 public class StorageContainerChannel {
 
     private final long scId;
@@ -139,8 +139,11 @@ private void fetchStorageContainerInfo() {
     }
 
     private void handleFetchStorageContainerInfoFailure(Throwable cause) {
-        log.info("Failed to fetch info of storage container ({}) - '{}'. Retry in {} ms ...",
-            new Object[]{scId, cause.getMessage(), ClientConstants.DEFAULT_BACKOFF_START_MS});
+        log.info()
+            .attr("scId", scId)
+            .exceptionMessage(cause)
+            .attr("retryMs", ClientConstants.DEFAULT_BACKOFF_START_MS)
+            .log("Failed to fetch info of storage container. Retrying ...");
         executor.schedule(() -> {
             fetchStorageContainerInfo();
         }, ClientConstants.DEFAULT_BACKOFF_START_MS, TimeUnit.MILLISECONDS);
@@ -182,8 +185,11 @@ private void handleFetchStorageContainerInfoSuccess(
         // get the channel from channel manager (if it doesn't exist create one)
         StorageServerChannel serverChannel = channelManager.getOrCreateChannel(endpoint.getRwEndpoint());
         if (null == serverChannel) {
-            log.info("No channel found/created for range server {}. The channel manager must be shutting down."
-                + " Stop the process of fetching storage container ({}).", endpoint.getRwEndpoint(), scId);
+            log.info()
+                .attr("rwEndpoint", endpoint.getRwEndpoint())
+                .attr("scId", scId)
+                .log("No channel found/created for range server. The channel manager must be shutting down."
+                    + " Stop the process of fetching storage container.");
             synchronized (this) {
                 rsChannelFuture.completeExceptionally(
                     new ObjectClosedException("StorageServerChannelManager is closed"));