Skip to content

KAFKA-9066: Retain metrics for failed tasks #8502

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rhauch merged 3 commits into from
Jun 11, 2020
Merged

KAFKA-9066: Retain metrics for failed tasks #8502

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b7e4f19
KAFKA-9066: Retain metrics for failed tasks
C0urante Apr 16, 2020
d191a22
KAFKA-9066: Address review feedback
C0urante Apr 17, 2020
2300698
KAFKA-9066: Fix unit tests
C0urante May 29, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -789,13 +789,18 @@ private void awaitStopTask(ConnectorTaskId taskId, long timeout) {
return;
}

connectorStatusMetricsGroup.recordTaskRemoved(taskId);
if (!task.awaitStop(timeout)) {
log.error("Graceful stop of task {} failed.", task.id());
task.cancel();
} else {
log.debug("Graceful stop of task {} succeeded.", task.id());
}

try {
task.removeMetrics();
} finally {
connectorStatusMetricsGroup.recordTaskRemoved(taskId);
}
}
}

Expand Down
8 changes: 6 additions & 2 deletions connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -175,8 +175,12 @@ protected void close() {
}

@Override
protected void releaseResources() {
sinkTaskMetricsGroup.close();
public void removeMetrics() {
try {
sinkTaskMetricsGroup.close();
} finally {
super.removeMetrics();
}
}

@Override
Expand Down
8 changes: 6 additions & 2 deletions connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,8 +188,12 @@ protected void close() {
}

@Override
protected void releaseResources() {
sourceTaskMetricsGroup.close();
public void removeMetrics() {
try {
sourceTaskMetricsGroup.close();
} finally {
super.removeMetrics();
}
}

@Override
Expand Down
27 changes: 10 additions & 17 deletions connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,16 +144,17 @@ public boolean awaitStop(long timeoutMs) {
}
}

/**
* Remove all metrics published by this task.
*/
public void removeMetrics() {
taskMetricsGroup.close();
}

protected abstract void execute();

protected abstract void close();

/**
* Method called when this worker task has been completely closed, and when the subclass should clean up
* all resources.
*/
protected abstract void releaseResources();

protected boolean isStopping() {
return stopping;
}
Expand Down Expand Up @@ -239,17 +240,9 @@ public void run() {
if (t instanceof Error)
throw (Error) t;
} finally {
try {
Thread.currentThread().setName(savedName);
Plugins.compareAndSwapLoaders(savedLoader);
shutdownLatch.countDown();
} finally {
try {
releaseResources();
} finally {
taskMetricsGroup.close();
}
}
Thread.currentThread().setName(savedName);
Plugins.compareAndSwapLoaders(savedLoader);
shutdownLatch.countDown();
}
}
}
Expand Down
9 changes: 0 additions & 9 deletions connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,9 +109,6 @@ public void standardStartup() {
workerTask.close();
expectLastCall();

workerTask.releaseResources();
EasyMock.expectLastCall();

statusListener.onShutdown(taskId);
expectLastCall();

Expand Down Expand Up @@ -153,9 +150,6 @@ public void stopBeforeStarting() {
workerTask.close();
EasyMock.expectLastCall();

workerTask.releaseResources();
EasyMock.expectLastCall();

replay(workerTask);

workerTask.initialize(TASK_CONFIG);
Expand Down Expand Up @@ -219,9 +213,6 @@ public Void answer() throws Throwable {
workerTask.close();
expectLastCall();

workerTask.releaseResources();
EasyMock.expectLastCall();

// there should be no call to onShutdown()

replay(workerTask);
Expand Down
12 changes: 12 additions & 0 deletions connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -602,6 +602,9 @@ public void testAddRemoveTask() throws Exception {
EasyMock.expect(workerTask.awaitStop(EasyMock.anyLong())).andStubReturn(true);
EasyMock.expectLastCall();

workerTask.removeMetrics();
EasyMock.expectLastCall();

expectStopStorage();
expectClusterId();

Expand Down Expand Up @@ -677,6 +680,9 @@ public void testTaskStatusMetricsStatuses() throws Exception {
EasyMock.expect(workerTask.awaitStop(EasyMock.anyLong())).andStubReturn(true);
EasyMock.expectLastCall();

workerTask.removeMetrics();
EasyMock.expectLastCall();

// Each time we check the task metrics, the worker will call the herder
herder.taskStatus(TASK_ID);
EasyMock.expectLastCall()
Expand Down Expand Up @@ -890,6 +896,9 @@ public void testCleanupTasksOnStop() throws Exception {
// Note that in this case we *do not* commit offsets since it's an unclean shutdown
EasyMock.expectLastCall();

workerTask.removeMetrics();
EasyMock.expectLastCall();

expectStopStorage();
expectClusterId();

Expand Down Expand Up @@ -964,6 +973,9 @@ public void testConverterOverrides() throws Exception {
EasyMock.expect(workerTask.awaitStop(EasyMock.anyLong())).andStubReturn(true);
EasyMock.expectLastCall();

workerTask.removeMetrics();
EasyMock.expectLastCall();

expectStopStorage();
expectClusterId();

Expand Down
12 changes: 12 additions & 0 deletions ...t/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerWithTopicCreationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -594,6 +594,9 @@ public void testAddRemoveTask() throws Exception {
EasyMock.expect(workerTask.awaitStop(EasyMock.anyLong())).andStubReturn(true);
EasyMock.expectLastCall();

workerTask.removeMetrics();
EasyMock.expectLastCall();

expectStopStorage();
expectClusterId();

Expand Down Expand Up @@ -669,6 +672,9 @@ public void testTaskStatusMetricsStatuses() throws Exception {
EasyMock.expect(workerTask.awaitStop(EasyMock.anyLong())).andStubReturn(true);
EasyMock.expectLastCall();

workerTask.removeMetrics();
EasyMock.expectLastCall();

// Each time we check the task metrics, the worker will call the herder
herder.taskStatus(TASK_ID);
EasyMock.expectLastCall()
Expand Down Expand Up @@ -879,6 +885,9 @@ public void testCleanupTasksOnStop() throws Exception {
// Note that in this case we *do not* commit offsets since it's an unclean shutdown
EasyMock.expectLastCall();

workerTask.removeMetrics();
EasyMock.expectLastCall();

expectStopStorage();
expectClusterId();

Expand Down Expand Up @@ -953,6 +962,9 @@ public void testConverterOverrides() throws Exception {
EasyMock.expect(workerTask.awaitStop(EasyMock.anyLong())).andStubReturn(true);
EasyMock.expectLastCall();

workerTask.removeMetrics();
EasyMock.expectLastCall();

expectStopStorage();
expectClusterId();

Expand Down