Skip to content

Commit 51abbc2

Browse files
merlimatmerlimat
authored
BP-69: Convert stream/distributedlog modules to slog (phase 5) (#4757)
* BP-69 base: add slog dependency and LICENSE entries Minimal scaffolding commit for the BP-69 slog migration series: - Add `io.github.merlimat.slog:slog:0.9.7` to the root pom.xml dependencyManagement and to the global compile classpath alongside the existing SLF4J API (which stays as the rendering backend). - Add lombok.config at the repo root so `@CustomLog` generates a slog `Logger` instead of an SLF4J one. - Register the slog jar in LICENSE-all.bin.txt, LICENSE-server.bin.txt and LICENSE-bkctl.bin.txt so the CI check-binary-license script finds it accounted for in the bundled-jars list. No actual Java file is converted in this commit. Individual module migrations stack on top of this one. * BP-69: Convert stream module from SLF4J to slog * Bump slog dependency to 0.9.8 Pulls in the MDC propagation fix from merlimat/slog#6, which makes log4j2 ThreadContext entries visible on slog events emitted via the Log4j2Logger backend (so %X{key} layouts and appenders that read event.getContextData().getValue(key) see the caller's MDC).
1 parent a3ec40b commit 51abbc2

213 files changed

Lines changed: 2151 additions & 1890 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

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

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@
2424
import io.grpc.NameResolverProvider;
2525
import java.net.URI;
2626
import javax.annotation.Nullable;
27-
import lombok.extern.slf4j.Slf4j;
27+
import lombok.CustomLog;
2828
import org.apache.bookkeeper.common.net.ServiceURI;
2929
import org.apache.bookkeeper.common.resolver.NameResolverFactoryProvider;
3030
import org.apache.bookkeeper.common.resolver.NameResolverProviderFactory;
@@ -42,7 +42,7 @@
4242
// https://github.com/grpc/grpc-java/issues/7133#issuecomment-680981331
4343
// Skipping the migration for now, will have to deal with this later when GRPC team
4444
// finalizes their API.
45-
@Slf4j
45+
@CustomLog
4646
public class BKRegistrationNameResolverProvider extends NameResolverFactoryProvider {
4747

4848
@Override
@@ -63,8 +63,10 @@ public NameResolver newNameResolver(URI targetUri, NameResolver.Args args) {
6363
serviceURI = ServiceURI.create(targetUri);
6464
} catch (NullPointerException | IllegalArgumentException e) {
6565
// invalid uri here, so return null to allow grpc to use other name resolvers
66-
log.info("BKRegistrationNameResolverProvider doesn't know how to resolve {} : cause {}",
67-
targetUri, e.getMessage());
66+
log.info()
67+
.attr("targetUri", targetUri)
68+
.exceptionMessage(e)
69+
.log("BKRegistrationNameResolverProvider doesn't know how to resolve target URI");
6870
return null;
6971
}
7072

@@ -73,7 +75,10 @@ public NameResolver newNameResolver(URI targetUri, NameResolver.Args args) {
7375
clientDriver = MetadataDrivers.getClientDriver(serviceURI.getUri());
7476
return new BKRegistrationNameResolver(clientDriver, serviceURI.getUri());
7577
} catch (IllegalArgumentException iae) {
76-
log.error("Unknown service uri : {}", serviceURI, iae);
78+
log.error()
79+
.attr("serviceURI", serviceURI)
80+
.exception(iae)
81+
.log("Unknown service uri");
7782
return null;
7883
}
7984
}

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

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@
2828
import io.netty.buffer.ByteBuf;
2929
import java.util.Optional;
3030
import java.util.concurrent.CompletableFuture;
31-
import lombok.extern.slf4j.Slf4j;
31+
import lombok.CustomLog;
3232
import org.apache.bookkeeper.api.StorageClient;
3333
import org.apache.bookkeeper.api.exceptions.ApiException;
3434
import org.apache.bookkeeper.api.kv.PTable;
@@ -50,7 +50,7 @@
5050
/**
5151
* The implementation of {@link StorageClient} client.
5252
*/
53-
@Slf4j
53+
@CustomLog
5454
public class SimpleStorageClientImpl extends SimpleClientBase implements StorageClient {
5555

5656
private static final String COMPONENT_NAME = SimpleStorageClientImpl.class.getSimpleName();
@@ -117,8 +117,11 @@ private void openTableImpl(String namespaceName,
117117
).thenCompose(resp -> {
118118
if (StatusCode.SUCCESS == resp.getCode()) {
119119
StreamProperties streamProps = resp.getStreamProps();
120-
log.info("Retrieved table properties for table {}/{} : {}",
121-
namespaceName, streamName, streamProps);
120+
log.info()
121+
.attr("namespace", namespaceName)
122+
.attr("streamName", streamName)
123+
.attr("streamProps", streamProps)
124+
.log("Retrieved table properties for table");
122125
if (StorageType.TABLE != streamProps.getStreamConf().getStorageType()) {
123126
return FutureUtils.exception(new ApiException(
124127
"Can't open a non-table storage entity : " + streamProps.getStreamConf().getStorageType()));

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

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
import io.netty.buffer.ByteBuf;
2323
import java.util.concurrent.CompletableFuture;
2424
import java.util.concurrent.TimeUnit;
25-
import lombok.extern.slf4j.Slf4j;
25+
import lombok.CustomLog;
2626
import org.apache.bookkeeper.api.StorageClient;
2727
import org.apache.bookkeeper.api.exceptions.ApiException;
2828
import org.apache.bookkeeper.api.kv.PTable;
@@ -44,7 +44,7 @@
4444
/**
4545
* The implementation of {@link StorageClient} client.
4646
*/
47-
@Slf4j
47+
@CustomLog
4848
public class StorageClientImpl extends AbstractAutoAsyncCloseable implements StorageClient {
4949

5050
private static final String COMPONENT_NAME = StorageClientImpl.class.getSimpleName();
@@ -122,9 +122,11 @@ private void openTableImpl(String namespaceName,
122122
CompletableFuture<PTable<ByteBuf, ByteBuf>> future) {
123123
FutureUtils.proxyTo(
124124
getStreamProperties(namespaceName, tableName).thenComposeAsync(props -> {
125-
if (log.isInfoEnabled()) {
126-
log.info("Retrieved table properties for table {}/{} : {}", namespaceName, tableName, props);
127-
}
125+
log.info()
126+
.attr("namespace", namespaceName)
127+
.attr("tableName", tableName)
128+
.attr("props", props)
129+
.log("Retrieved table properties for table");
128130
if (StorageType.TABLE != props.getStreamConf().getStorageType()) {
129131
return FutureUtils.exception(new ApiException(
130132
"Can't open a non-table storage entity : " + props.getStreamConf().getStorageType())
@@ -167,7 +169,7 @@ public void close() {
167169
try {
168170
super.close(1, TimeUnit.SECONDS);
169171
} catch (Exception e) {
170-
log.warn("Encountered exceptions on closing the storage client", e);
172+
log.warn().exception(e).log("Encountered exceptions on closing the storage client");
171173
}
172174
scheduler.forceShutdown(100, TimeUnit.MILLISECONDS);
173175
}

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

Lines changed: 4 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
import java.util.concurrent.CompletableFuture;
2222
import java.util.concurrent.ExecutionException;
2323
import java.util.function.Supplier;
24-
import lombok.extern.slf4j.Slf4j;
24+
import lombok.CustomLog;
2525
import org.apache.bookkeeper.api.StorageClient;
2626
import org.apache.bookkeeper.clients.StorageClientImpl;
2727
import org.apache.bookkeeper.clients.config.StorageClientSettings;
@@ -38,7 +38,7 @@
3838
/**
3939
* A storage admin client.
4040
*/
41-
@Slf4j
41+
@CustomLog
4242
public class StorageAdminClientImpl extends AbstractAutoAsyncCloseable implements StorageAdminClient {
4343

4444
// clients
@@ -127,13 +127,9 @@ public void close() {
127127
try {
128128
closeAsync().get();
129129
} catch (InterruptedException e) {
130-
if (log.isDebugEnabled()) {
131-
log.debug("Interrupted on closing stream admin client", e);
132-
}
130+
log.debug().exception(e).log("Interrupted on closing stream admin client");
133131
} catch (ExecutionException e) {
134-
if (log.isDebugEnabled()) {
135-
log.debug("Failed to cloe stream admin client", e);
136-
}
132+
log.debug().exception(e).log("Failed to cloe stream admin client");
137133
}
138134
}
139135
}

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@
3131
import static org.mockito.Mockito.when;
3232

3333
import io.netty.buffer.ByteBuf;
34-
import lombok.extern.slf4j.Slf4j;
34+
import lombok.CustomLog;
3535
import org.apache.bookkeeper.api.exceptions.ApiException;
3636
import org.apache.bookkeeper.api.kv.PTable;
3737
import org.apache.bookkeeper.api.kv.Table;
@@ -51,7 +51,7 @@
5151
* Unit test {@link StorageClientImpl}.
5252
*/
5353
@RunWith(MockitoJUnitRunner.class)
54-
@Slf4j
54+
@CustomLog
5555
public class StorageClientImplTest extends GrpcClientTestBase {
5656

5757
private static final String NAMESPACE = "test-namespace";

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

Lines changed: 17 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -24,14 +24,14 @@
2424
import java.util.concurrent.locks.ReentrantReadWriteLock;
2525
import java.util.function.Function;
2626
import javax.annotation.Nullable;
27-
import lombok.extern.slf4j.Slf4j;
27+
import lombok.CustomLog;
2828
import org.apache.bookkeeper.clients.config.StorageClientSettings;
2929
import org.apache.bookkeeper.stream.proto.common.Endpoint;
3030

3131
/**
3232
* A manager manages channels to range servers.
3333
*/
34-
@Slf4j
34+
@CustomLog
3535
public class StorageServerChannelManager implements AutoCloseable {
3636

3737
private final ReentrantReadWriteLock lock;
@@ -69,21 +69,21 @@ public boolean addStorageServer(Endpoint endpoint, StorageServerChannel channel)
6969
lock.readLock().lock();
7070
try {
7171
if (closed) {
72-
log.warn("Skip adding channel {} of range server {} since the channel manager is already closed",
73-
channel, endpoint);
72+
log.warn()
73+
.attr("channel", channel)
74+
.attr("endpoint", endpoint)
75+
.log("Skip adding channel since the channel manager is already closed");
7476
channel.close();
7577
return false;
7678
}
7779

7880
StorageServerChannel oldChannel = channels.putIfAbsent(endpoint, channel);
7981
if (null != oldChannel) {
80-
if (log.isDebugEnabled()) {
81-
log.debug("KeyRange server ({}) already existed in the channel manager.", endpoint);
82-
}
82+
log.debug().attr("endpoint", endpoint).log("KeyRange server already existed in the channel manager.");
8383
channel.close();
8484
return false;
8585
} else {
86-
log.info("Added range server ({}) into the channel manager.", endpoint);
86+
log.info().attr("endpoint", endpoint).log("Added range server into the channel manager.");
8787
return true;
8888
}
8989
} finally {
@@ -117,8 +117,10 @@ public StorageServerChannel removeChannel(Endpoint endpoint, StorageServerChanne
117117
lock.readLock().lock();
118118
try {
119119
if (closed) {
120-
log.warn("Skip removing channel {} of range server {} since the channel manager is already closed",
121-
channel, endpoint);
120+
log.warn()
121+
.attr("channel", channel)
122+
.attr("endpoint", endpoint)
123+
.log("Skip removing channel since the channel manager is already closed");
122124
return null;
123125
}
124126

@@ -133,12 +135,12 @@ public StorageServerChannel removeChannel(Endpoint endpoint, StorageServerChanne
133135
}
134136
}
135137
if (null == channelRemoved) {
136-
if (log.isDebugEnabled()) {
137-
log.debug("No channel associated with endpoint {} to be removed.", endpoint);
138-
}
138+
log.debug().attr("endpoint", endpoint).log("No channel associated with endpoint to be removed.");
139139
} else {
140-
log.info("Removed channel {} for range server {} successfully",
141-
channelRemoved, endpoint);
140+
log.info()
141+
.attr("channel", channelRemoved)
142+
.attr("endpoint", endpoint)
143+
.log("Removed channel for range server successfully");
142144
}
143145
if (null != channelRemoved) {
144146
channelRemoved.close();

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

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
import java.util.concurrent.ScheduledExecutorService;
2626
import java.util.concurrent.TimeUnit;
2727
import javax.annotation.concurrent.GuardedBy;
28-
import lombok.extern.slf4j.Slf4j;
28+
import lombok.CustomLog;
2929
import org.apache.bookkeeper.clients.exceptions.StorageContainerException;
3030
import org.apache.bookkeeper.clients.impl.channel.StorageServerChannel;
3131
import org.apache.bookkeeper.clients.impl.channel.StorageServerChannelManager;
@@ -43,7 +43,7 @@
4343
/**
4444
* A client place holder for managing information of storage containers.
4545
*/
46-
@Slf4j
46+
@CustomLog
4747
public class StorageContainerChannel {
4848

4949
private final long scId;
@@ -139,8 +139,11 @@ private void fetchStorageContainerInfo() {
139139
}
140140

141141
private void handleFetchStorageContainerInfoFailure(Throwable cause) {
142-
log.info("Failed to fetch info of storage container ({}) - '{}'. Retry in {} ms ...",
143-
new Object[]{scId, cause.getMessage(), ClientConstants.DEFAULT_BACKOFF_START_MS});
142+
log.info()
143+
.attr("scId", scId)
144+
.exceptionMessage(cause)
145+
.attr("retryMs", ClientConstants.DEFAULT_BACKOFF_START_MS)
146+
.log("Failed to fetch info of storage container. Retrying ...");
144147
executor.schedule(() -> {
145148
fetchStorageContainerInfo();
146149
}, ClientConstants.DEFAULT_BACKOFF_START_MS, TimeUnit.MILLISECONDS);
@@ -182,8 +185,11 @@ private void handleFetchStorageContainerInfoSuccess(
182185
// get the channel from channel manager (if it doesn't exist create one)
183186
StorageServerChannel serverChannel = channelManager.getOrCreateChannel(endpoint.getRwEndpoint());
184187
if (null == serverChannel) {
185-
log.info("No channel found/created for range server {}. The channel manager must be shutting down."
186-
+ " Stop the process of fetching storage container ({}).", endpoint.getRwEndpoint(), scId);
188+
log.info()
189+
.attr("rwEndpoint", endpoint.getRwEndpoint())
190+
.attr("scId", scId)
191+
.log("No channel found/created for range server. The channel manager must be shutting down."
192+
+ " Stop the process of fetching storage container.");
187193
synchronized (this) {
188194
rsChannelFuture.completeExceptionally(
189195
new ObjectClosedException("StorageServerChannelManager is closed"));

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

Lines changed: 9 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -27,13 +27,13 @@
2727
import io.grpc.ClientInterceptors.CheckedForwardingClientCall;
2828
import io.grpc.Metadata;
2929
import io.grpc.MethodDescriptor;
30-
import lombok.extern.slf4j.Slf4j;
30+
import lombok.CustomLog;
3131
import org.apache.bookkeeper.common.grpc.netty.LongBinaryMarshaller;
3232

3333
/**
3434
* A client interceptor that intercepting outgoing calls to storage containers.
3535
*/
36-
@Slf4j
36+
@CustomLog
3737
public class StorageContainerClientInterceptor implements ClientInterceptor {
3838

3939
private final long scId;
@@ -50,28 +50,23 @@ public StorageContainerClientInterceptor(long scId) {
5050
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method,
5151
CallOptions callOptions,
5252
Channel next) {
53-
if (log.isTraceEnabled()) {
54-
log.trace("Intercepting method {} : req marshaller = {}, resp marshaller = {}",
55-
method.getFullMethodName(),
56-
method.getRequestMarshaller(),
57-
method.getResponseMarshaller());
58-
}
53+
log.trace()
54+
.attr("method", method.getFullMethodName())
55+
.attr("reqMarshaller", method.getRequestMarshaller())
56+
.attr("respMarshaller", method.getResponseMarshaller())
57+
.log("Intercepting method");
5958
return new CheckedForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
6059
@Override
6160
protected void checkedStart(Listener<RespT> responseListener,
6261
Metadata headers) throws Exception {
63-
if (log.isTraceEnabled()) {
64-
log.trace("Attaching storage container {},", scId);
65-
}
62+
log.trace().attr("scId", scId).log("Attaching storage container");
6663
headers.put(scIdKey, scId);
6764
delegate().start(responseListener, headers);
6865
}
6966

7067
@Override
7168
public void request(int numMessages) {
72-
if (log.isTraceEnabled()) {
73-
log.trace("request {} messages", numMessages);
74-
}
69+
log.trace().attr("numMessages", numMessages).log("request messages");
7570
super.request(numMessages);
7671
}
7772
};

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

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,12 +22,12 @@
2222
import java.util.Map;
2323
import javax.annotation.Nullable;
2424
import javax.annotation.concurrent.GuardedBy;
25-
import lombok.extern.slf4j.Slf4j;
25+
import lombok.CustomLog;
2626

2727
/**
2828
* A manager manages the mapping between storage containers and range servers.
2929
*/
30-
@Slf4j
30+
@CustomLog
3131
public class StorageContainerManager implements AutoCloseable {
3232

3333
@GuardedBy("storageContainers")
@@ -48,7 +48,8 @@ public boolean replaceStorageContainer(long groupId, StorageContainerInfo groupI
4848
synchronized (storageContainers) {
4949
StorageContainerInfo oldGroupInfo = storageContainers.get(groupId);
5050
if (null == oldGroupInfo || oldGroupInfo.getRevision() < groupInfo.getRevision()) {
51-
log.info("Updated the storage container info for group {} : {}", groupId, groupInfo);
51+
log.info().attr("groupId", groupId).attr("groupInfo", groupInfo)
52+
.log("Updated the storage container info");
5253
storageContainers.put(groupId, groupInfo);
5354
return true;
5455
}

‎stream/clients/java/base/src/main/java/org/apache/bookkeeper/clients/impl/internal/LocationClientImpl.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@
3232
import java.util.concurrent.CompletableFuture;
3333
import java.util.function.Predicate;
3434
import java.util.stream.Stream;
35-
import lombok.extern.slf4j.Slf4j;
35+
import lombok.CustomLog;
3636
import org.apache.bookkeeper.clients.config.StorageClientSettings;
3737
import org.apache.bookkeeper.clients.impl.internal.api.LocationClient;
3838
import org.apache.bookkeeper.clients.utils.ClientConstants;
@@ -50,7 +50,7 @@
5050
/**
5151
* Default Implementation of {@link LocationClient}.
5252
*/
53-
@Slf4j
53+
@CustomLog
5454
public class LocationClientImpl implements LocationClient {
5555

5656
private final StorageClientSettings settings;
@@ -83,7 +83,7 @@ private Stream<Long> getDefaultBackoffs() {
8383
cause -> shouldRetryOnException(cause);
8484

8585
private static boolean shouldRetryOnException(Throwable cause) {
86-
log.error("Not able to locate storage container ", cause);
86+
log.error().exception(cause).log("Not able to locate storage container");
8787
if (cause instanceof StatusRuntimeException || cause instanceof StatusException) {
8888
Status status;
8989
if (cause instanceof StatusException) {

0 commit comments

Comments
 (0)