kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ewe...@apache.org
Subject kafka git commit: MINOR: Standardize logging of Worker-level messages from Tasks and Connectors
Date Wed, 09 Aug 2017 16:07:37 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 1a653c813 -> a593db6a2


MINOR: Standardize logging of Worker-level messages from Tasks and Connectors

This ensures all logs have the connector/task ID, whether tasks are source or sink, and formats
them consistently.

Author: Ewen Cheslack-Postava <me@ewencp.org>

Reviewers: Randall Hauch <rhauch@gmail.com>, Konstantine Karantasis <konstantine@confluent.io>,
Jason Gustafson <jason@confluent.io>

Closes #3639 from ewencp/standardize-connector-task-logging


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/a593db6a
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/a593db6a
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/a593db6a

Branch: refs/heads/trunk
Commit: a593db6a2b6c3511215008b6d3dd6bd77f84e8c4
Parents: 1a653c8
Author: Ewen Cheslack-Postava <me@ewencp.org>
Authored: Wed Aug 9 09:07:27 2017 -0700
Committer: Ewen Cheslack-Postava <me@ewencp.org>
Committed: Wed Aug 9 09:07:27 2017 -0700

----------------------------------------------------------------------
 .../runtime/SourceTaskOffsetCommitter.java      |  6 +--
 .../kafka/connect/runtime/WorkerConnector.java  | 19 ++++---
 .../kafka/connect/runtime/WorkerSinkTask.java   | 54 ++++++++++----------
 .../kafka/connect/runtime/WorkerSourceTask.java | 46 ++++++++---------
 .../kafka/connect/runtime/WorkerTask.java       |  6 +--
 .../connect/runtime/WorkerSinkTaskTest.java     |  2 +
 6 files changed, 66 insertions(+), 67 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/a593db6a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitter.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitter.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitter.java
index f93d7d2..c502809 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitter.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitter.java
@@ -103,17 +103,17 @@ class SourceTaskOffsetCommitter {
     }
 
     private void commit(WorkerSourceTask workerTask) {
-        log.debug("Committing offsets for {}", workerTask);
+        log.debug("{} Committing offsets", workerTask);
         try {
             if (workerTask.commitOffsets()) {
                 return;
             }
-            log.error("Failed to commit offsets for {}", workerTask);
+            log.error("{} Failed to commit offsets", workerTask);
         } catch (Throwable t) {
             // We're very careful about exceptions here since any uncaught exceptions in
the commit
             // thread would cause the fixed interval schedule on the ExecutorService to stop
running
             // for that task
-            log.error("Unhandled exception when committing {}: ", workerTask, t);
+            log.error("{} Unhandled exception when committing: ", workerTask, t);
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/a593db6a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java
index 1937429..983db92 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java
@@ -68,7 +68,7 @@ public class WorkerConnector {
     public void initialize(ConnectorConfig connectorConfig) {
         try {
             this.config = connectorConfig.originalsStrings();
-            log.debug("Initializing connector {} with config {}", connName, config);
+            log.debug("{} Initializing connector {} with config {}", this, connName, config);
 
             connector.initialize(new ConnectorContext() {
                 @Override
@@ -78,13 +78,13 @@ public class WorkerConnector {
 
                 @Override
                 public void raiseError(Exception e) {
-                    log.error("Connector raised an error {}", connName, e);
+                    log.error("{} Connector raised an error", this, e);
                     onFailure(e);
                     ctx.raiseError(e);
                 }
             });
         } catch (Throwable t) {
-            log.error("Error initializing connector {}", connName, t);
+            log.error("{} Error initializing connector", this, t);
             onFailure(t);
         }
     }
@@ -105,7 +105,7 @@ public class WorkerConnector {
                     throw new IllegalArgumentException("Cannot start connector in state "
+ state);
             }
         } catch (Throwable t) {
-            log.error("Error while starting connector {}", connName, t);
+            log.error("{} Error while starting connector", this, t);
             onFailure(t);
             return false;
         }
@@ -149,7 +149,7 @@ public class WorkerConnector {
                     throw new IllegalArgumentException("Cannot pause connector in state "
+ state);
             }
         } catch (Throwable t) {
-            log.error("Error while shutting down connector {}", connName, t);
+            log.error("{} Error while shutting down connector", this, t);
             statusListener.onFailure(connName, t);
             this.state = State.FAILED;
         }
@@ -161,7 +161,7 @@ public class WorkerConnector {
                 connector.stop();
             this.state = State.STOPPED;
         } catch (Throwable t) {
-            log.error("Error while shutting down connector {}", connName, t);
+            log.error("{} Error while shutting down connector", this, t);
             this.state = State.FAILED;
         } finally {
             statusListener.onShutdown(connName);
@@ -170,11 +170,11 @@ public class WorkerConnector {
 
     public void transitionTo(TargetState targetState) {
         if (state == State.FAILED) {
-            log.warn("Cannot transition connector {} to {} since it has failed", connName,
targetState);
+            log.warn("{} Cannot transition connector to {} since it has failed", this, targetState);
             return;
         }
 
-        log.debug("Transition connector {} to {}", connName, targetState);
+        log.debug("{} Transition connector to {}", this, targetState);
         if (targetState == TargetState.PAUSED) {
             pause();
         } else if (targetState == TargetState.STARTED) {
@@ -198,8 +198,7 @@ public class WorkerConnector {
     @Override
     public String toString() {
         return "WorkerConnector{" +
-                "connName='" + connName + '\'' +
-                ", connector=" + connector +
+                "id=" + connName +
                 '}';
     }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/a593db6a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
index c4567a3..9295712 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
@@ -112,7 +112,7 @@ class WorkerSinkTask extends WorkerTask {
             this.consumer = createConsumer();
             this.context = new WorkerSinkTaskContext(consumer);
         } catch (Throwable t) {
-            log.error("Task {} failed initialization and will not be started.", t);
+            log.error("{} Task failed initialization and will not be started.", this, t);
             onFailure(t);
         }
     }
@@ -170,7 +170,7 @@ class WorkerSinkTask extends WorkerTask {
 
             // Check for timed out commits
             if (committing && now >= commitTimeoutMs) {
-                log.warn("Commit of {} offsets timed out", this);
+                log.warn("{} Commit of offsets timed out", this);
                 commitFailures++;
                 committing = false;
             }
@@ -179,7 +179,7 @@ class WorkerSinkTask extends WorkerTask {
             long timeoutMs = Math.max(nextCommit - now, 0);
             poll(timeoutMs);
         } catch (WakeupException we) {
-            log.trace("{} consumer woken up", id);
+            log.trace("{} consumer woken up", this);
 
             if (isStopping())
                 return;
@@ -197,15 +197,13 @@ class WorkerSinkTask extends WorkerTask {
 
     private void onCommitCompleted(Throwable error, long seqno) {
         if (commitSeqno != seqno) {
-            log.debug("Got callback for timed out commit {}: {}, but most recent commit is
{}",
-                    this,
-                    seqno, commitSeqno);
+            log.debug("{} Got callback for timed out commit: {}, but most recent commit is
{}", this, seqno, commitSeqno);
         } else {
             if (error != null) {
-                log.error("Commit of {} offsets threw an unexpected exception: ", this, error);
+                log.error("{} Commit of offsets threw an unexpected exception: ", this, error);
                 commitFailures++;
             } else {
-                log.debug("Finished {} offset commit successfully in {} ms",
+                log.debug("{} Finished offset commit successfully in {} ms",
                         this, time.milliseconds() - commitStarted);
                 commitFailures = 0;
             }
@@ -221,16 +219,16 @@ class WorkerSinkTask extends WorkerTask {
      * Initializes and starts the SinkTask.
      */
     protected void initializeAndStart() {
-        log.debug("Initializing task {} ", id);
+        log.debug("{} Initializing task", this);
         String topicsStr = taskConfig.get(SinkTask.TOPICS_CONFIG);
         if (topicsStr == null || topicsStr.isEmpty())
             throw new ConnectException("Sink tasks require a list of topics.");
         String[] topics = topicsStr.split(",");
-        log.debug("Task {} subscribing to topics {}", id, topics);
+        log.debug("{} Task subscribing to topics {}", this, topics);
         consumer.subscribe(Arrays.asList(topics), new HandleRebalance());
         task.initialize(context);
         task.start(taskConfig);
-        log.info("Sink task {} finished initialization and start", this);
+        log.info("{} Sink task finished initialization and start", this);
     }
 
     /** Poll for new messages with the given timeout. Should only be invoked by the worker
thread. */
@@ -242,10 +240,10 @@ class WorkerSinkTask extends WorkerTask {
             context.timeout(-1L);
         }
 
-        log.trace("{} polling consumer with timeout {} ms", id, timeoutMs);
+        log.trace("{} polling consumer with timeout {} ms", this, timeoutMs);
         ConsumerRecords<byte[], byte[]> msgs = pollConsumer(timeoutMs);
         assert messageBatch.isEmpty() || msgs.isEmpty();
-        log.trace("{} polling returned {} messages", id, msgs.count());
+        log.trace("{} polling returned {} messages", this, msgs.count());
 
         convertMessages(msgs);
         deliverMessages();
@@ -305,12 +303,12 @@ class WorkerSinkTask extends WorkerTask {
             taskProvidedOffsets = task.preCommit(new HashMap<>(currentOffsets));
         } catch (Throwable t) {
             if (closing) {
-                log.warn("{} Offset commit failed during close");
+                log.warn("{} Offset commit failed during close", this);
                 onCommitCompleted(t, commitSeqno);
             } else {
                 log.error("{} Offset commit failed, rewinding to last committed offsets",
this, t);
                 for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : lastCommittedOffsets.entrySet())
{
-                    log.debug("{} Rewinding topic partition {} to offset {}", id, entry.getKey(),
entry.getValue().offset());
+                    log.debug("{} Rewinding topic partition {} to offset {}", this, entry.getKey(),
entry.getValue().offset());
                     consumer.seek(entry.getKey(), entry.getValue().offset());
                 }
                 currentOffsets = new HashMap<>(lastCommittedOffsets);
@@ -334,13 +332,17 @@ class WorkerSinkTask extends WorkerTask {
             final TopicPartition partition = taskProvidedOffsetEntry.getKey();
             final OffsetAndMetadata taskProvidedOffset = taskProvidedOffsetEntry.getValue();
             if (commitableOffsets.containsKey(partition)) {
-                if (taskProvidedOffset.offset() <= currentOffsets.get(partition).offset())
{
+                long taskOffset = taskProvidedOffset.offset();
+                long currentOffset = currentOffsets.get(partition).offset();
+                if (taskOffset <= currentOffset) {
                     commitableOffsets.put(partition, taskProvidedOffset);
                 } else {
-                    log.warn("Ignoring invalid task provided offset {}/{} -- not yet consumed",
partition, taskProvidedOffset);
+                    log.warn("{} Ignoring invalid task provided offset {}/{} -- not yet consumed,
taskOffset={} currentOffset={}",
+                            this, partition, taskProvidedOffset, taskOffset, currentOffset);
                 }
             } else {
-                log.warn("Ignoring invalid task provided offset {}/{} -- partition not assigned",
partition, taskProvidedOffset);
+                log.warn("{} Ignoring invalid task provided offset {}/{} -- partition not
assigned, assignment={}",
+                        this, partition, taskProvidedOffset, consumer.assignment());
             }
         }
 
@@ -402,7 +404,7 @@ class WorkerSinkTask extends WorkerTask {
 
     private void convertMessages(ConsumerRecords<byte[], byte[]> msgs) {
         for (ConsumerRecord<byte[], byte[]> msg : msgs) {
-            log.trace("Consuming message with key {}, value {}", msg.key(), msg.value());
+            log.trace("{} Consuming message with key {}, value {}", this, msg.key(), msg.value());
             SchemaAndValue keyAndSchema = keyConverter.toConnectData(msg.topic(), msg.key());
             SchemaAndValue valueAndSchema = valueConverter.toConnectData(msg.topic(), msg.value());
             SinkRecord record = new SinkRecord(msg.topic(), msg.partition(),
@@ -445,16 +447,16 @@ class WorkerSinkTask extends WorkerTask {
                 pausedForRedelivery = false;
             }
         } catch (RetriableException e) {
-            log.error("RetriableException from SinkTask {}:", id, e);
+            log.error("{} RetriableException from SinkTask:", this, e);
             // If we're retrying a previous batch, make sure we've paused all topic partitions
so we don't get new data,
             // but will still be able to poll in order to handle user-requested timeouts,
keep group membership, etc.
             pausedForRedelivery = true;
             pauseAll();
             // Let this exit normally, the batch will be reprocessed on the next loop.
         } catch (Throwable t) {
-            log.error("Task {} threw an uncaught and unrecoverable exception", id, t);
-            log.error("Task is being killed and will not recover until manually restarted");
-            throw new ConnectException("Exiting WorkerSinkTask due to unrecoverable exception.");
+            log.error("{} Task threw an uncaught and unrecoverable exception. Task is being
killed and will not "
+                    + "recover until manually restarted.", this, t);
+            throw new ConnectException("Exiting WorkerSinkTask due to unrecoverable exception.",
t);
         }
     }
 
@@ -467,12 +469,12 @@ class WorkerSinkTask extends WorkerTask {
             TopicPartition tp = entry.getKey();
             Long offset = entry.getValue();
             if (offset != null) {
-                log.trace("Rewind {} to offset {}.", tp, offset);
+                log.trace("{} Rewind {} to offset {}.", this, tp, offset);
                 consumer.seek(tp, offset);
                 lastCommittedOffsets.put(tp, new OffsetAndMetadata(offset));
                 currentOffsets.put(tp, new OffsetAndMetadata(offset));
             } else {
-                log.warn("Cannot rewind {} to null offset.", tp);
+                log.warn("{} Cannot rewind {} to null offset.", this, tp);
             }
         }
         context.clearOffsets();
@@ -495,7 +497,7 @@ class WorkerSinkTask extends WorkerTask {
                 long pos = consumer.position(tp);
                 lastCommittedOffsets.put(tp, new OffsetAndMetadata(pos));
                 currentOffsets.put(tp, new OffsetAndMetadata(pos));
-                log.debug("{} assigned topic partition {} with offset {}", id, tp, pos);
+                log.debug("{} assigned topic partition {} with offset {}", this, tp, pos);
             }
 
             // If we paused everything for redelivery (which is no longer relevant since
we discarded the data), make

http://git-wip-us.apache.org/repos/asf/kafka/blob/a593db6a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
index e676d50..0f42186 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
@@ -113,7 +113,7 @@ class WorkerSourceTask extends WorkerTask {
         try {
             this.taskConfig = taskConfig.originalsStrings();
         } catch (Throwable t) {
-            log.error("Task {} failed initialization and will not be started.", t);
+            log.error("{} Task failed initialization and will not be started.", this, t);
             onFailure(t);
         }
     }
@@ -140,7 +140,7 @@ class WorkerSourceTask extends WorkerTask {
         try {
             task.initialize(new WorkerSourceTaskContext(offsetReader));
             task.start(taskConfig);
-            log.info("Source task {} finished initialization and start", this);
+            log.info("{} Source task finished initialization and start", this);
             synchronized (this) {
                 if (startedShutdownBeforeStartCompleted) {
                     task.stop();
@@ -159,12 +159,12 @@ class WorkerSourceTask extends WorkerTask {
                 }
 
                 if (toSend == null) {
-                    log.debug("Nothing to send to Kafka. Polling source for additional records");
+                    log.debug("{} Nothing to send to Kafka. Polling source for additional
records", this);
                     toSend = task.poll();
                 }
                 if (toSend == null)
                     continue;
-                log.debug("About to send " + toSend.size() + " records to Kafka");
+                log.debug("{} About to send " + toSend.size() + " records to Kafka", this);
                 if (!sendRecords())
                     stopRequestedLatch.await(SEND_FAILED_BACKOFF_MS, TimeUnit.MILLISECONDS);
             }
@@ -198,7 +198,7 @@ class WorkerSourceTask extends WorkerTask {
             byte[] value = valueConverter.fromConnectData(record.topic(), record.valueSchema(),
record.value());
             final ProducerRecord<byte[], byte[]> producerRecord = new ProducerRecord<>(record.topic(),
record.kafkaPartition(),
                     ConnectUtils.checkAndConvertTimestamp(record.timestamp()), key, value);
-            log.trace("Appending record with key {}, value {}", record.key(), record.value());
+            log.trace("{} Appending record with key {}, value {}", this, record.key(), record.value());
             // We need this queued first since the callback could happen immediately (even
synchronously in some cases).
             // Because of this we need to be careful about handling retries -- we always
save the previously attempted
             // record as part of toSend and need to use a flag to track whether we should
actually add it to the outstanding
@@ -227,10 +227,11 @@ class WorkerSourceTask extends WorkerTask {
                                     // timeouts, callbacks with exceptions should never be
invoked in practice. If the
                                     // user overrode these settings, the best we can do is
notify them of the failure via
                                     // logging.
-                                    log.error("{} failed to send record to {}: {}", id, topic,
e);
-                                    log.debug("Failed record: {}", preTransformRecord);
+                                    log.error("{} failed to send record to {}: {}", this,
topic, e);
+                                    log.debug("{} Failed record: {}", this, preTransformRecord);
                                 } else {
-                                    log.trace("Wrote record successfully: topic {} partition
{} offset {}",
+                                    log.trace("{} Wrote record successfully: topic {} partition
{} offset {}",
+                                            this,
                                             recordMetadata.topic(), recordMetadata.partition(),
                                             recordMetadata.offset());
                                     commitTaskRecord(preTransformRecord);
@@ -240,7 +241,7 @@ class WorkerSourceTask extends WorkerTask {
                         });
                 lastSendFailed = false;
             } catch (RetriableException e) {
-                log.warn("Failed to send {}, backing off before retrying:", producerRecord,
e);
+                log.warn("{} Failed to send {}, backing off before retrying:", this, producerRecord,
e);
                 toSend = toSend.subList(processed, toSend.size());
                 lastSendFailed = true;
                 return false;
@@ -256,10 +257,8 @@ class WorkerSourceTask extends WorkerTask {
     private void commitTaskRecord(SourceRecord record) {
         try {
             task.commitRecord(record);
-        } catch (InterruptedException e) {
-            log.error("Exception thrown", e);
         } catch (Throwable t) {
-            log.error("Exception thrown while calling task.commitRecord()", t);
+            log.error("{} Exception thrown while calling task.commitRecord()", this, t);
         }
     }
 
@@ -270,8 +269,7 @@ class WorkerSourceTask extends WorkerTask {
             removed = outstandingMessagesBacklog.remove(record);
         // But if neither one had it, something is very wrong
         if (removed == null) {
-            log.error("CRITICAL Saw callback for record that was not present in the outstanding
message set: "
-                    + "{}", record);
+            log.error("{} CRITICAL Saw callback for record that was not present in the outstanding
message set: {}", this, record);
         } else if (flushing && outstandingMessages.isEmpty()) {
             // flush thread may be waiting on the outstanding messages to clear
             this.notifyAll();
@@ -303,7 +301,7 @@ class WorkerSourceTask extends WorkerTask {
                 try {
                     long timeoutMs = timeout - time.milliseconds();
                     if (timeoutMs <= 0) {
-                        log.error("Failed to flush {}, timed out while waiting for producer
to flush outstanding {} messages", this, outstandingMessages.size());
+                        log.error("{} Failed to flush, timed out while waiting for producer
to flush outstanding {} messages", this, outstandingMessages.size());
                         finishFailedFlush();
                         return false;
                     }
@@ -324,7 +322,7 @@ class WorkerSourceTask extends WorkerTask {
                 // flush time, which can be used for monitoring even if the connector doesn't
record any
                 // offsets.
                 finishSuccessfulFlush();
-                log.debug("Finished {} offset commitOffsets successfully in {} ms",
+                log.debug("{} Finished offset commitOffsets successfully in {} ms",
                         this, time.milliseconds() - started);
 
                 commitSourceTask();
@@ -337,9 +335,9 @@ class WorkerSourceTask extends WorkerTask {
             @Override
             public void onCompletion(Throwable error, Void result) {
                 if (error != null) {
-                    log.error("Failed to flush {} offsets to storage: ", this, error);
+                    log.error("{} Failed to flush offsets to storage: ", this, error);
                 } else {
-                    log.trace("Finished flushing {} offsets to storage", this);
+                    log.trace("{} Finished flushing offsets to storage", this);
                 }
             }
         });
@@ -356,21 +354,21 @@ class WorkerSourceTask extends WorkerTask {
             // errors, is only wasteful in this minor edge case, and the worst result is
that the log
             // could look a little confusing.
         } catch (InterruptedException e) {
-            log.warn("Flush of {} offsets interrupted, cancelling", this);
+            log.warn("{} Flush of offsets interrupted, cancelling", this);
             finishFailedFlush();
             return false;
         } catch (ExecutionException e) {
-            log.error("Flush of {} offsets threw an unexpected exception: ", this, e);
+            log.error("{} Flush of offsets threw an unexpected exception: ", this, e);
             finishFailedFlush();
             return false;
         } catch (TimeoutException e) {
-            log.error("Timed out waiting to flush {} offsets to storage", this);
+            log.error("{} Timed out waiting to flush offsets to storage", this);
             finishFailedFlush();
             return false;
         }
 
         finishSuccessfulFlush();
-        log.info("Finished {} commitOffsets successfully in {} ms",
+        log.info("{} Finished commitOffsets successfully in {} ms",
                 this, time.milliseconds() - started);
 
         commitSourceTask();
@@ -381,10 +379,8 @@ class WorkerSourceTask extends WorkerTask {
     private void commitSourceTask() {
         try {
             this.task.commit();
-        } catch (InterruptedException ex) {
-            log.warn("Commit interrupted", ex);
         } catch (Throwable t) {
-            log.error("Exception thrown while calling task.commit()", t);
+            log.error("{} Exception thrown while calling task.commit()", this, t);
         }
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/a593db6a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java
index 9b233dd..3295434 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java
@@ -124,7 +124,7 @@ abstract class WorkerTask implements Runnable {
         try {
             close();
         } catch (Throwable t) {
-            log.error("Task {} threw an uncaught and unrecoverable exception during shutdown",
id, t);
+            log.error("{} Task threw an uncaught and unrecoverable exception during shutdown",
this, t);
             throw t;
         }
     }
@@ -145,8 +145,8 @@ abstract class WorkerTask implements Runnable {
 
             execute();
         } catch (Throwable t) {
-            log.error("Task {} threw an uncaught and unrecoverable exception", id, t);
-            log.error("Task is being killed and will not recover until manually restarted");
+            log.error("{} Task threw an uncaught and unrecoverable exception", this, t);
+            log.error("{} Task is being killed and will not recover until manually restarted",
this);
             throw t;
         } finally {
             doClose();

http://git-wip-us.apache.org/repos/asf/kafka/blob/a593db6a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java
b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java
index eae3726..50b7503 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java
@@ -461,6 +461,8 @@ public class WorkerSinkTaskTest {
 
         sinkTask.preCommit(workerCurrentOffsets);
         EasyMock.expectLastCall().andReturn(taskOffsets);
+        // Expect extra invalid topic partition to be filtered, which causes the consumer
assignment to be logged
+        EasyMock.expect(consumer.assignment()).andReturn(workerCurrentOffsets.keySet());
         final Capture<OffsetCommitCallback> callback = EasyMock.newCapture();
         consumer.commitAsync(EasyMock.eq(committableOffsets), EasyMock.capture(callback));
         EasyMock.expectLastCall().andAnswer(new IAnswer<Void>() {


Mime
View raw message