KAFKA-5763: Refactor NetworkClient to use LogContext #3761

adyach · 2017-08-30T19:40:27Z

This PR lets logging client id in every log line in NetworkClient

ijuma · 2017-08-30T19:51:44Z

Thanks for the PR. It would be good to receive the LogContext from the calling client (Consumer, AdminClient, Producer). Also, the log messages that include a client id explicitly should be changed.

adyach · 2017-08-30T20:02:30Z

@ijuma If we pass LogContext to constructor, then NetworkClient will log with the prefix of the LogContext, which could be [Consumer clientId=" + clientId + ", groupId=" + groupId + "]. Is that what you expect to see? I have not found any explicit logs with client id in NetworkClient.

ijuma · 2017-08-30T20:13:37Z

Yes, the idea is that one retains the context of the client within the networking classes. The log statement will include the actual logging class separately.

hachikuji · 2017-08-30T21:09:08Z

Should we consider pushing the log context into Selector as well?

ijuma

@hachikuji, yes, I was thinking the same thing regarding the Selector. A couple more comments inline.

ijuma · 2017-08-31T09:31:12Z

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -327,21 +329,22 @@ static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcesso
    static KafkaAdminClient createInternal(AdminClientConfig config, KafkaClient client, Metadata metadata, Time time) {
        Metrics metrics = null;
        String clientId = generateClientId(config);
+        LogContext logContext = new LogContext("[KafkaAdminClient clientId=" + clientId + "] ");


Maybe we should extract a static method createLogContext to avoid duplicating the logic. Also, AdminClient (instead of KafkaAdminClient) is shorter and provides enough context.

@ijuma 0847add

ijuma · 2017-08-31T09:37:51Z

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

@@ -160,7 +161,8 @@ public NetworkClient(Selectable selector,
        this(metadataUpdater, null, selector, clientId, maxInFlightRequestsPerConnection,
             reconnectBackoffMs, reconnectBackoffMax,
             socketSendBuffer, socketReceiveBuffer, requestTimeoutMs, time,
-             discoverBrokerVersions, apiVersions, null);
+             discoverBrokerVersions, apiVersions, null,
+                new LogContext("[NetworkClient clientId=" + clientId + "] "));


Maybe we can receive a LogContext in this constructor as well. This constructor is typically used for inter-broker communication and the callers can provide a useful context too.

@ijuma 0847add

omkreddy · 2017-09-01T05:35:58Z

core/src/main/scala/kafka/server/KafkaServer.scala

@@ -365,6 +365,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP

    def doControlledShutdown(retries: Int): Boolean = {
      val metadataUpdater = new ManualMetadataUpdater()
+      val logContext = new LogContext("[KafkaServer clientId=" + config.brokerId.toString + "] ")


Should we use "borkerId=" here?

I thought since NetworkClient uses it as clientId, it made sense to use clientId, but you are right it could be misleading
99a77c2

omkreddy · 2017-09-01T05:40:48Z

core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala

@@ -46,7 +44,7 @@ object TransactionMarkerChannelManager {
            txnStateManager: TransactionStateManager,
            txnMarkerPurgatory: DelayedOperationPurgatory[DelayedTxnMarker],
            time: Time): TransactionMarkerChannelManager = {
-
+    val logContext = new LogContext("[TransactionMarkerChannelManager clientId=" + config.brokerId + "] ")


Should we use "borkerId=" here?

omkreddy · 2017-09-01T05:41:24Z

core/src/main/scala/kafka/controller/ControllerChannelManager.scala

@@ -108,6 +108,7 @@ class ControllerChannelManager(controllerContext: ControllerContext, config: Kaf
    val messageQueue = new LinkedBlockingQueue[QueueItem]
    debug("Controller %d trying to connect to broker %d".format(config.brokerId, broker.id))
    val brokerNode = broker.getNode(config.interBrokerListenerName)
+    val logContext = new LogContext("[ControllerChannelManager clientId=" + config.brokerId.toString + "] ")


Should we use "borkerId=" here?

ijuma

Thanks for the updates, I think we're close. A few more comments.

ijuma · 2017-09-01T08:35:52Z

clients/src/main/java/org/apache/kafka/common/network/Selector.java

@@ -114,26 +114,27 @@

    /**
     * Create a new nioSelector
-     *
-     * @param maxReceiveSize Max size in bytes of a single network receive (use {@link NetworkReceive#UNLIMITED} for no limit)
+     *  @param maxReceiveSize Max size in bytes of a single network receive (use {@link NetworkReceive#UNLIMITED} for no limit)


This was probably unintentional.

ijuma · 2017-09-01T08:43:28Z

core/src/main/scala/kafka/controller/ControllerChannelManager.scala

@@ -108,6 +108,7 @@ class ControllerChannelManager(controllerContext: ControllerContext, config: Kaf
    val messageQueue = new LinkedBlockingQueue[QueueItem]
    debug("Controller %d trying to connect to broker %d".format(config.brokerId, broker.id))
    val brokerNode = broker.getNode(config.interBrokerListenerName)
+    val logContext = new LogContext("[ControllerChannelManager brokerId=" + config.brokerId.toString + "] ")


Maybe this could be:

s"[ControllerChannelManager controllerId=${config.brokerId}, targetBrokerId=${brokerNode.idString}] "

It's a bit long though, we could potentially shorten ControllerChannelManager to Controller and replace controllerId with id. @hachikuji, what do you think?

Yeah, I'd vote to shorten both. Seems not much room for confusion.

ijuma · 2017-09-01T08:44:25Z

core/src/main/scala/kafka/controller/ControllerChannelManager.scala

@@ -125,7 +126,8 @@ class ControllerChannelManager(controllerContext: ControllerContext, config: Kaf
        "controller-channel",
        Map("broker-id" -> broker.id.toString).asJava,


We should take the chance and replace broker.id.toString with broker.idString.

ijuma · 2017-09-01T08:48:49Z

core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala

@@ -46,7 +44,7 @@ object TransactionMarkerChannelManager {
            txnStateManager: TransactionStateManager,
            txnMarkerPurgatory: DelayedOperationPurgatory[DelayedTxnMarker],
            time: Time): TransactionMarkerChannelManager = {
-
+    val logContext = new LogContext("[TransactionMarkerChannelManager brokerId=" + config.brokerId + "] ")


We could pass this to TransactionMarkerChannelManager so that it can be used to set logIdent (instead of duplicating the logic). We'd have to add a logPrefix method to LogContext.

@ijuma should I also replace kafka.utils.Logging usage which uses logIdent with LogContext?

I suggest not doing that in this PR. We can open a separate JIRA about that to discuss the best way to do it. My suggestion was to use the same log prefix for logIdent and logContext and the easiest way to do that is to create the LogContext first and then pass its value to logIdent. Does that make sense?

Yes, it makes total sense to move it to another ticket.

ijuma · 2017-09-01T08:52:27Z

core/src/main/scala/kafka/network/SocketServer.scala

@@ -441,7 +441,8 @@ private[kafka] class Processor(val id: Int,
    false,
    true,
    ChannelBuilders.serverChannelBuilder(listenerName, securityProtocol, config, credentialProvider.credentialCache),
-    memoryPool)
+    memoryPool,
+    new LogContext())


In SocketServer, we have:

this.logIdent = "[Socket Server on Broker " + config.brokerId + "], "

We could set the logContext as a field in SocketServer, use it to set logIdent and then pass it here. We can probably use [SocketServer brokerId=${config.brokerId}] as the prefix.

ijuma · 2017-09-01T08:56:17Z

core/src/main/scala/kafka/server/KafkaServer.scala

@@ -365,6 +365,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP

    def doControlledShutdown(retries: Int): Boolean = {
      val metadataUpdater = new ManualMetadataUpdater()
+      val logContext = new LogContext("[KafkaServer brokerId=" + config.brokerId.toString + "] ")


Again, we may want to do this at the same time as logIdent to keep things consistent.

ijuma · 2017-09-01T08:58:36Z

core/src/main/scala/kafka/server/ReplicaFetcherBlockingSend.scala

@@ -49,6 +49,7 @@ class ReplicaFetcherBlockingSend(sourceBroker: BrokerEndPoint,
  private val socketTimeout: Int = brokerConfig.replicaSocketTimeoutMs

  private val networkClient = {
+    val logContext = new LogContext("[ReplicaFetcherBlockingSend clientId=" + clientId + "] ")


I would pass this from ReplicaFetcherThread. We can take the chance and set a reasonable logIdent there too.

adyach · 2017-09-04T15:06:28Z

@ijuma I have fixed review comments, could you give it another look?

ijuma

Thanks, just a few more tweaks and I think we're done.

ijuma · 2017-09-05T13:36:31Z

core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala


-    new TransactionCoordinator(config.brokerId, scheduler, producerIdManager, txnStateManager, txnMarkerChannelManager, time)
+    val logContext = new LogContext("[Transaction Coordinator " + config.brokerId + "] ")


To maintain the idea of key/value pairs in a context, we can maybe change this to: new LogContext(s"[TransactionCoordinator id=${config.brokerId}] ")

ijuma · 2017-09-05T13:38:06Z

core/src/main/scala/kafka/server/KafkaServer.scala

@@ -198,7 +200,8 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP
        /* generate brokerId */
        val (brokerId, initialOfflineDirs) = getBrokerIdAndOfflineDirs
        config.brokerId = brokerId
-        this.logIdent = "[Kafka Server " + config.brokerId + "], "
+        logContext = new LogContext("[Kafka Server " + config.brokerId + "] ")


Similarly, this can be changed to new LogContext("[KafkaServer id=${config.brokerId}] ")

ijuma · 2017-09-05T13:51:17Z

core/src/main/scala/kafka/server/ReplicaFetcherThread.scala

@@ -54,8 +54,10 @@ class ReplicaFetcherThread(name: String,
  type REQ = FetchRequest
  type PD = PartitionData

+  private val logContext = new LogContext(s"[ReplicaFetcherThread brokerId=${brokerConfig.brokerId}] ")


This could be s"[ReplicaFetcher brokerId=${brokerConfig.brokerId}, sourceBrokerId=${sourceBroker.id}, fetcherId=$fetcherId] ") and we can then remove mentions of the source broker from individual log statements.

adyach · 2017-09-06T16:09:55Z

@ijuma i have addressed your comments

adyach · 2017-09-07T19:28:13Z

@ijuma ping

ijuma · 2017-09-08T02:58:55Z

@adyach No need to ping within such a short period of time. We have a lot of things going on in parallel. Can you please review and, if you agree, integrate the following changes into your PR?

ijuma@7f1ae22

Thanks!

ijuma · 2017-09-11T08:04:18Z

Thanks for integrating the change. Looks like there's a minor conflict in SocketServer. Can you resolve that, please?

adyach · 2017-09-11T10:33:54Z

@ijuma resolved the conflicts, thank you for the help

ijuma

Thanks for the updates and for your patience. LGTM. This should make it easier to debug issues when they happen. :) Merging to trunk.

adyach added 2 commits August 30, 2017 21:38

kafka-5763: LogConotext to creatre logger with prefix

4b1b42f

This PR lets logging client id in every log line in NetworkClient

kafka-5763: NetworkClient vs Consumer

482318b

adyach added 3 commits August 30, 2017 23:00

kafka-5763: pass LogContext to NetworkClient

d2d1666

kafka-5763: reverted imports change

dabf664

kafka-5763: removed unused import

04bb4b4

ijuma reviewed Aug 31, 2017

View reviewed changes

adyach added 3 commits August 31, 2017 20:07

KAFKA-5763: LogContext to all NetworkClient ctrs

0847add

KAFKA-5763: LogContext to Selector ctrs

e3e550e

KAFKA-5763: fixed scala compilation issues

459217d

omkreddy reviewed Sep 1, 2017

View reviewed changes

KAFKA-5763: brokerId vs clientId

99a77c2

ijuma reviewed Sep 1, 2017

View reviewed changes

adyach added 2 commits September 3, 2017 21:07

KAFKA-5763: moved LogContext to contructor

76925e2

KAFKA-5763: fixed test data contruction

d9da6a0

ijuma reviewed Sep 5, 2017

View reviewed changes

KAFKA-5763: fixed namings

c1863b9

Minor improvements

9ec277b

Merge remote-tracking branch 'upstream/trunk' into kafka-5763

c94e107

adyach added 2 commits September 11, 2017 10:29

KAFKA-5763: fixed ctr params in test

56ed463

KAFKA-5763: fixed ctr params in test

97f8692

ijuma approved these changes Sep 11, 2017

View reviewed changes

asfgit closed this in 9bb06c3 Sep 11, 2017

		@@ -125,7 +126,8 @@ class ControllerChannelManager(controllerContext: ControllerContext, config: Kaf
		"controller-channel",
		Map("broker-id" -> broker.id.toString).asJava,


		new TransactionCoordinator(config.brokerId, scheduler, producerIdManager, txnStateManager, txnMarkerChannelManager, time)
		val logContext = new LogContext("[Transaction Coordinator " + config.brokerId + "] ")

KAFKA-5763: Refactor NetworkClient to use LogContext #3761

KAFKA-5763: Refactor NetworkClient to use LogContext #3761

Conversation

adyach commented Aug 30, 2017

ijuma commented Aug 30, 2017

adyach commented Aug 30, 2017

ijuma commented Aug 30, 2017

hachikuji commented Aug 30, 2017

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

omkreddy Sep 1, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adyach Sep 2, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adyach commented Sep 4, 2017

ijuma left a comment

Choose a reason for hiding this comment

ijuma Sep 5, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adyach commented Sep 6, 2017

adyach commented Sep 7, 2017

ijuma commented Sep 8, 2017

ijuma commented Sep 11, 2017

adyach commented Sep 11, 2017

ijuma left a comment

Choose a reason for hiding this comment

omkreddy Sep 1, 2017 •

edited

Loading

adyach Sep 2, 2017 •

edited

Loading

ijuma Sep 5, 2017 •

edited

Loading