KAFKA-5111: code cleanup follow up #2917

mjsax · 2017-04-26T03:29:56Z

mainly moving methods
also improved logging

mjsax · 2017-04-26T03:31:34Z

Call for review @enothereska @dguy @guozhangwang

This is a follow-up for #2895 with the code cleanup commit that I did remove (moving around methods). I also put a second commit on top to improving logging. This PR also addresses as remaining nits from #2895

dguy

couple minor comments, otherwise LGTM

dguy · 2017-04-26T09:41:29Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java

@@ -122,14 +105,14 @@ public void commit() {
     */
    @Override
    public void suspend() {
-        log.info("{} Suspending", logPrefix);
+        log.debug("{} Suspending", logPrefix);


any reason why we are changing these from info to debug?

We log suspended tasks at info level in StreamThread already.

Seems we are not actually?

for (final AbstractTask task : activeAndStandbytasks()) { + try { + task.suspend(); + } catch (final RuntimeException e) { + firstException.compareAndSet(null, e); + } + }

We do. It's little further up:

log.debug("{} suspendTasksAndState: suspending all active tasks {} and standby tasks {}", logPrefix, activeTasks.keySet(), standbyTasks.keySet()); final AtomicReference<RuntimeException> firstException = new AtomicReference<>(null); for (final AbstractTask task : activeAndStandbytasks()) { try { task.suspend(); } catch (final RuntimeException e) { firstException.compareAndSet(null, e); } }

@mjsax Just realized we already have an info log entry in onPartitionXX callbacks, so this should be fine. NVM.

dguy · 2017-04-26T09:43:30Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

+        final Sensor taskCommitTimeSensor;
+
+
+        public TaskMetrics(final StreamsMetrics metrics) {


nit: can be package private

asfbot · 2017-04-26T17:19:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3182/
Test PASSed (JDK 8 and Scala 2.11).

mjsax · 2017-04-26T18:24:25Z

Updated this.

asfbot · 2017-04-26T19:19:07Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3182/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-04-26T20:19:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3177/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-04-27T05:19:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3197/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-04-27T05:19:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3199/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-04-27T14:19:07Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3212/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-04-27T14:19:07Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3212/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-04-27T15:13:52Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3220/
Test PASSed (JDK 8 and Scala 2.11).

guozhangwang · 2017-04-27T21:35:41Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java

    /**
     * Flush all state stores owned by this task
     */
    void flushState() {
+        log.trace("{} Flushing state store", logPrefix);


I feel it is a bit misleading since it is not flushing a store, but the manager.

BTW, maybe it's better just to move

log.debug("{} Flushing all stores registered in the state manager", logPrefix);

out of the if condition inside manager.flush() and avoid adding this?

Well, it's that the same "thing" the manager is just an implementation details?
The idea is, to log that task.flushState() was called -- this log is independent of "how" the flushing is done. Does this make sense?

And I thinks it better to keep it within the if -> this los is not about a method got called, but what does actually happen -- and thus only log if an actual flush is done.

guozhangwang · 2017-04-27T21:38:05Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java

+     * @param writeCheckpoint boolean indicating if a checkpoint file should be written
+     */
+    void closeStateManager(final boolean writeCheckpoint) throws ProcessorStateException {
+        log.trace("{} Closing state stores", logPrefix);


Ditto to my other comment, maybe we can just move

log.debug("{} Closing its state manager and all the registered state stores", logPrefix);

out of the if condition inside the manager code.

guozhangwang · 2017-04-27T21:40:37Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java

+    }
+
+    protected void updateOffsetLimits() {
+        log.trace("{} Updating store offset limits", logPrefix);


Change trace to debug?

I'll change this to debug and do the inner as trace

Just trying to be consistent with other task-level log entries. We can also just remove this as you mentioned below.

guozhangwang · 2017-04-27T21:44:00Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

-        initTopology();
+    @Override
+    protected void flushState() {
+        log.trace("{} Flushing state and producer topology", logPrefix);


producer topology -> producer?

guozhangwang · 2017-04-27T21:48:18Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

+    @Override
+    protected void flushState() {
+        log.trace("{} Flushing state and producer topology", logPrefix);
+        super.flushState();


It will incur two trace logging one from stream-task and one from abstract-task if trace is enabled.

Instead could we just add a trace log after line 262 for "Flushing the producer after all state store has been flushed and corresponding changelog records sent".

same reply as above -- all those log indicate that a method got called -- thus it's not "double logging" imho -- but it's of course verbose, but for trace level should be fine

guozhangwang · 2017-04-27T21:49:56Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java

+        for (final TopicPartition partition : partitions) {
+            try {
+                final OffsetAndMetadata metadata = consumer.committed(partition); // TODO: batch API?
+                stateMgr.putOffsetLimit(partition, metadata != null ? metadata.offset() : 0L);


Add a trace entry recording the updated offset limit inside the manager class?

guozhangwang · 2017-04-27T21:53:16Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

        streamsMetrics = new StreamsMetricsThreadImpl(metrics, "stream-metrics", "thread." + threadClientId,
            Collections.singletonMap("client-id", threadClientId));
        if (config.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG) < 0) {
-            log.warn("Negative cache size passed in thread [{}]. Reverting to cache size of 0 bytes.", threadClientId);
+            log.warn("{} Negative cache size passed in thread [{}]. Reverting to cache size of 0 bytes.", logPrefix, threadClientId);


The second param seems redundant.

guozhangwang · 2017-04-27T21:54:53Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -682,7 +781,7 @@ private void maybeUpdateStandbyTasks() {
                    final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> remainingStandbyRecords = new HashMap<>();

                    for (final Map.Entry<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> entry :
-                            standbyRecords.entrySet()) {
+                        standbyRecords.entrySet()) {


This newline is not needed.

guozhangwang · 2017-04-27T21:59:15Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java

@@ -122,14 +105,14 @@ public void commit() {
     */
    @Override
    public void suspend() {
-        log.info("{} Suspending", logPrefix);
+        log.debug("{} Suspending", logPrefix);


Seems we are not actually?

for (final AbstractTask task : activeAndStandbytasks()) { + try { + task.suspend(); + } catch (final RuntimeException e) { + firstException.compareAndSet(null, e); + } + }

guozhangwang · 2017-04-27T22:05:46Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -939,6 +1209,7 @@ private void addStreamTasks(final Collection<TopicPartition> assignment, final l
        final Map<TaskId, Set<TopicPartition>> newTasks = new HashMap<>();

        // collect newly assigned tasks and reopen re-assigned tasks
+        log.info("{} Resuming suspended re-assigned tasks", logPrefix);


This seems a bit misleading, and less informative. Maybe just "Adding active tasks {partitionAssignor.activeTasks()}"?

Ditto below for standby tasks

Why? Cf Line 1221 -> task.resume()
This loop figures out what resume task got reassigned.

guozhangwang · 2017-04-27T22:06:43Z

retest this please

asfbot · 2017-04-27T22:29:19Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3236/
Test FAILed (JDK 7 and Scala 2.10).

guozhangwang · 2017-04-27T22:50:10Z

@mjsax is Jenkins failure relevant?

asfbot · 2017-04-27T22:58:32Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3241/
Test FAILed (JDK 8 and Scala 2.11).

mjsax · 2017-04-27T23:03:30Z

It says: :clients:findbugsMain FAILED

I'll try to rebase.

mjsax · 2017-04-27T23:20:35Z

Updated and rebased. \cc @guozhangwang

asfbot · 2017-04-27T23:21:47Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3232/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-04-27T23:41:04Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3246/
Test FAILed (JDK 7 and Scala 2.10).

guozhangwang · 2017-04-28T00:12:13Z

Merged to trunk.

asfbot · 2017-04-28T00:53:29Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3249/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-04-28T00:54:00Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3245/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-04-28T02:57:41Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3254/
Test FAILed (JDK 8 and Scala 2.11).

dguy reviewed Apr 26, 2017

View reviewed changes

mjsax changed the title ~~Kafka 5111 code cleanup follow up~~ KAFKA-5111: code cleanup follow up Apr 26, 2017

enothereska approved these changes Apr 27, 2017

View reviewed changes

guozhangwang reviewed Apr 27, 2017

View reviewed changes

mjsax added 4 commits April 27, 2017 16:18

KAFA-5111: Improve internal Task API (code cleanup follow up)

3369ad2

Improve logging

d8d7ebb

Damian's comment

4ce8247

Github comments

3f545d7

mjsax force-pushed the kafka-5111-code-cleanup-follow-up branch from cd6cbea to 3f545d7 Compare April 27, 2017 23:18

Github comments

c893df4

asfgit closed this in 6753af2 Apr 28, 2017

mjsax deleted the kafka-5111-code-cleanup-follow-up branch April 28, 2017 00:48

		final Sensor taskCommitTimeSensor;


		public TaskMetrics(final StreamsMetrics metrics) {

KAFKA-5111: code cleanup follow up #2917

KAFKA-5111: code cleanup follow up #2917

Conversation

mjsax commented Apr 26, 2017

mjsax commented Apr 26, 2017

dguy 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

Choose a reason for hiding this comment

asfbot commented Apr 26, 2017

mjsax commented Apr 26, 2017

asfbot commented Apr 26, 2017

asfbot commented Apr 26, 2017

asfbot commented Apr 27, 2017

asfbot commented Apr 27, 2017

asfbot commented Apr 27, 2017

asfbot commented Apr 27, 2017

asfbot commented Apr 27, 2017

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

mjsax Apr 27, 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

mjsax Apr 27, 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

guozhangwang commented Apr 27, 2017

asfbot commented Apr 27, 2017

guozhangwang commented Apr 27, 2017

asfbot commented Apr 27, 2017

mjsax commented Apr 27, 2017

mjsax commented Apr 27, 2017

asfbot commented Apr 27, 2017

asfbot commented Apr 27, 2017

guozhangwang commented Apr 28, 2017

asfbot commented Apr 28, 2017

asfbot commented Apr 28, 2017

asfbot commented Apr 28, 2017

mjsax Apr 27, 2017 •

edited

Loading

mjsax Apr 27, 2017 •

edited

Loading