Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
Forever310
druid
提交
09fcfc3b
D
druid
项目概览
Forever310
/
druid
与 Fork 源项目一致
从无法访问的项目Fork
通知
3
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
D
druid
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
09fcfc3b
编写于
7月 18, 2014
作者:
G
Gian Merlino
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
Fix race in RemoteTaskRunner that could lead to zombie tasks.
上级
9e303527
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
51 addition
and
33 deletion
+51
-33
indexing-service/src/main/java/io/druid/indexing/overlord/RemoteTaskRunner.java
...ain/java/io/druid/indexing/overlord/RemoteTaskRunner.java
+18
-18
indexing-service/src/main/java/io/druid/indexing/overlord/RemoteTaskRunnerWorkItem.java
.../io/druid/indexing/overlord/RemoteTaskRunnerWorkItem.java
+19
-1
indexing-service/src/test/java/io/druid/indexing/overlord/scaling/SimpleResourceManagementStrategyTest.java
...verlord/scaling/SimpleResourceManagementStrategyTest.java
+14
-14
未找到文件。
indexing-service/src/main/java/io/druid/indexing/overlord/RemoteTaskRunner.java
浏览文件 @
09fcfc3b
...
...
@@ -422,11 +422,7 @@ public class RemoteTaskRunner implements TaskRunner, TaskLogStreamer
private
RemoteTaskRunnerWorkItem
addPendingTask
(
final
Task
task
)
{
log
.
info
(
"Added pending task %s"
,
task
.
getId
());
final
RemoteTaskRunnerWorkItem
taskRunnerWorkItem
=
new
RemoteTaskRunnerWorkItem
(
task
.
getId
(),
SettableFuture
.<
TaskStatus
>
create
(),
null
);
final
RemoteTaskRunnerWorkItem
taskRunnerWorkItem
=
new
RemoteTaskRunnerWorkItem
(
task
.
getId
(),
null
);
pendingTaskPayloads
.
put
(
task
.
getId
(),
task
);
pendingTasks
.
put
(
task
.
getId
(),
taskRunnerWorkItem
);
runPendingTasks
();
...
...
@@ -663,17 +659,24 @@ public class RemoteTaskRunner implements TaskRunner, TaskLogStreamer
if
((
tmp
=
runningTasks
.
get
(
taskId
))
!=
null
)
{
taskRunnerWorkItem
=
tmp
;
}
else
{
log
.
warn
(
"Worker[%s] announced a status for a task I didn't know about, adding to runningTasks: %s"
,
zkWorker
.
getWorker
().
getHost
(),
taskId
);
taskRunnerWorkItem
=
new
RemoteTaskRunnerWorkItem
(
final
RemoteTaskRunnerWorkItem
newTaskRunnerWorkItem
=
new
RemoteTaskRunnerWorkItem
(
taskId
,
SettableFuture
.<
TaskStatus
>
create
(),
zkWorker
.
getWorker
()
);
runningTasks
.
put
(
taskId
,
taskRunnerWorkItem
.
withWorker
(
zkWorker
.
getWorker
()));
final
RemoteTaskRunnerWorkItem
existingItem
=
runningTasks
.
putIfAbsent
(
taskId
,
newTaskRunnerWorkItem
);
if
(
existingItem
==
null
)
{
log
.
warn
(
"Worker[%s] announced a status for a task I didn't know about, adding to runningTasks: %s"
,
zkWorker
.
getWorker
().
getHost
(),
taskId
);
taskRunnerWorkItem
=
newTaskRunnerWorkItem
;
}
else
{
taskRunnerWorkItem
=
existingItem
;
}
}
if
(
taskStatus
.
isComplete
())
{
...
...
@@ -835,7 +838,7 @@ public class RemoteTaskRunner implements TaskRunner, TaskLogStreamer
// Worker is done with this task
zkWorker
.
setLastCompletedTaskTime
(
new
DateTime
());
}
else
{
log
.
info
(
"
No worker run task[%s]
with status[%s]"
,
taskStatus
.
getId
(),
taskStatus
.
getStatusCode
());
log
.
info
(
"
Workerless task[%s] completed
with status[%s]"
,
taskStatus
.
getId
(),
taskStatus
.
getStatusCode
());
}
// Move from running -> complete
...
...
@@ -843,9 +846,6 @@ public class RemoteTaskRunner implements TaskRunner, TaskLogStreamer
runningTasks
.
remove
(
taskStatus
.
getId
());
// Notify interested parties
final
ListenableFuture
<
TaskStatus
>
result
=
taskRunnerWorkItem
.
getResult
();
if
(
result
!=
null
)
{
((
SettableFuture
<
TaskStatus
>)
result
).
set
(
taskStatus
);
}
taskRunnerWorkItem
.
setResult
(
taskStatus
);
}
}
indexing-service/src/main/java/io/druid/indexing/overlord/RemoteTaskRunnerWorkItem.java
浏览文件 @
09fcfc3b
...
...
@@ -32,6 +32,24 @@ public class RemoteTaskRunnerWorkItem extends TaskRunnerWorkItem
private
final
Worker
worker
;
public
RemoteTaskRunnerWorkItem
(
String
taskId
,
Worker
worker
)
{
this
(
taskId
,
SettableFuture
.<
TaskStatus
>
create
(),
worker
);
}
public
RemoteTaskRunnerWorkItem
(
String
taskId
,
DateTime
createdTime
,
DateTime
queueInsertionTime
,
Worker
worker
)
{
this
(
taskId
,
SettableFuture
.<
TaskStatus
>
create
(),
createdTime
,
queueInsertionTime
,
worker
);
}
private
RemoteTaskRunnerWorkItem
(
String
taskId
,
SettableFuture
<
TaskStatus
>
result
,
Worker
worker
...
...
@@ -42,7 +60,7 @@ public class RemoteTaskRunnerWorkItem extends TaskRunnerWorkItem
this
.
worker
=
worker
;
}
p
ublic
RemoteTaskRunnerWorkItem
(
p
rivate
RemoteTaskRunnerWorkItem
(
String
taskId
,
SettableFuture
<
TaskStatus
>
result
,
DateTime
createdTime
,
...
...
indexing-service/src/test/java/io/druid/indexing/overlord/scaling/SimpleResourceManagementStrategyTest.java
浏览文件 @
09fcfc3b
...
...
@@ -27,8 +27,8 @@ import com.metamx.emitter.EmittingLogger;
import
com.metamx.emitter.service.ServiceEmitter
;
import
com.metamx.emitter.service.ServiceEventBuilder
;
import
io.druid.common.guava.DSuppliers
;
import
io.druid.indexing.common.TestMergeTask
;
import
io.druid.indexing.common.TaskStatus
;
import
io.druid.indexing.common.TestMergeTask
;
import
io.druid.indexing.common.task.NoopTask
;
import
io.druid.indexing.common.task.Task
;
import
io.druid.indexing.overlord.RemoteTaskRunnerWorkItem
;
...
...
@@ -113,7 +113,7 @@ public class SimpleResourceManagementStrategyTest
boolean
provisionedSomething
=
simpleResourceManagementStrategy
.
doProvision
(
Arrays
.<
RemoteTaskRunnerWorkItem
>
asList
(
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
,
null
).
withQueueInsertionTime
(
new
DateTime
())
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
).
withQueueInsertionTime
(
new
DateTime
())
),
Arrays
.<
ZkWorker
>
asList
(
new
TestZkWorker
(
testTask
)
...
...
@@ -141,7 +141,7 @@ public class SimpleResourceManagementStrategyTest
boolean
provisionedSomething
=
simpleResourceManagementStrategy
.
doProvision
(
Arrays
.<
RemoteTaskRunnerWorkItem
>
asList
(
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
,
null
).
withQueueInsertionTime
(
new
DateTime
())
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
).
withQueueInsertionTime
(
new
DateTime
())
),
Arrays
.<
ZkWorker
>
asList
(
new
TestZkWorker
(
testTask
)
...
...
@@ -157,7 +157,7 @@ public class SimpleResourceManagementStrategyTest
provisionedSomething
=
simpleResourceManagementStrategy
.
doProvision
(
Arrays
.<
RemoteTaskRunnerWorkItem
>
asList
(
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
,
null
).
withQueueInsertionTime
(
new
DateTime
())
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
).
withQueueInsertionTime
(
new
DateTime
())
),
Arrays
.<
ZkWorker
>
asList
(
new
TestZkWorker
(
testTask
)
...
...
@@ -188,7 +188,7 @@ public class SimpleResourceManagementStrategyTest
EasyMock
.
expect
(
autoScalingStrategy
.
ipToIdLookup
(
EasyMock
.<
List
<
String
>>
anyObject
()))
.
andReturn
(
Lists
.<
String
>
newArrayList
()).
times
(
2
);
EasyMock
.
expect
(
autoScalingStrategy
.
terminateWithIds
(
EasyMock
.<
List
<
String
>>
anyObject
()))
.
andReturn
(
null
);
.
andReturn
(
null
);
EasyMock
.
expect
(
autoScalingStrategy
.
provision
()).
andReturn
(
new
AutoScalingData
(
Lists
.<
String
>
newArrayList
(
"fake"
))
);
...
...
@@ -196,7 +196,7 @@ public class SimpleResourceManagementStrategyTest
boolean
provisionedSomething
=
simpleResourceManagementStrategy
.
doProvision
(
Arrays
.<
RemoteTaskRunnerWorkItem
>
asList
(
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
,
null
).
withQueueInsertionTime
(
new
DateTime
())
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
).
withQueueInsertionTime
(
new
DateTime
())
),
Arrays
.<
ZkWorker
>
asList
(
new
TestZkWorker
(
testTask
)
...
...
@@ -214,7 +214,7 @@ public class SimpleResourceManagementStrategyTest
provisionedSomething
=
simpleResourceManagementStrategy
.
doProvision
(
Arrays
.<
RemoteTaskRunnerWorkItem
>
asList
(
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
,
null
).
withQueueInsertionTime
(
new
DateTime
())
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
).
withQueueInsertionTime
(
new
DateTime
())
),
Arrays
.<
ZkWorker
>
asList
(
new
TestZkWorker
(
testTask
)
...
...
@@ -248,7 +248,7 @@ public class SimpleResourceManagementStrategyTest
boolean
terminatedSomething
=
simpleResourceManagementStrategy
.
doTerminate
(
Arrays
.<
RemoteTaskRunnerWorkItem
>
asList
(
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
,
null
).
withQueueInsertionTime
(
new
DateTime
())
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
).
withQueueInsertionTime
(
new
DateTime
())
),
Arrays
.<
ZkWorker
>
asList
(
new
TestZkWorker
(
null
)
...
...
@@ -278,7 +278,7 @@ public class SimpleResourceManagementStrategyTest
boolean
terminatedSomething
=
simpleResourceManagementStrategy
.
doTerminate
(
Arrays
.<
RemoteTaskRunnerWorkItem
>
asList
(
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
,
null
).
withQueueInsertionTime
(
new
DateTime
())
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
).
withQueueInsertionTime
(
new
DateTime
())
),
Arrays
.<
ZkWorker
>
asList
(
new
TestZkWorker
(
null
)
...
...
@@ -293,7 +293,7 @@ public class SimpleResourceManagementStrategyTest
terminatedSomething
=
simpleResourceManagementStrategy
.
doTerminate
(
Arrays
.<
RemoteTaskRunnerWorkItem
>
asList
(
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
,
null
).
withQueueInsertionTime
(
new
DateTime
())
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
).
withQueueInsertionTime
(
new
DateTime
())
),
Arrays
.<
ZkWorker
>
asList
(
new
TestZkWorker
(
null
)
...
...
@@ -319,7 +319,7 @@ public class SimpleResourceManagementStrategyTest
boolean
terminatedSomething
=
simpleResourceManagementStrategy
.
doTerminate
(
Arrays
.<
RemoteTaskRunnerWorkItem
>
asList
(
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
,
null
).
withQueueInsertionTime
(
new
DateTime
())
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
).
withQueueInsertionTime
(
new
DateTime
())
),
Arrays
.<
ZkWorker
>
asList
(
new
TestZkWorker
(
NoopTask
.
create
()),
...
...
@@ -337,7 +337,7 @@ public class SimpleResourceManagementStrategyTest
boolean
provisionedSomething
=
simpleResourceManagementStrategy
.
doProvision
(
Arrays
.<
RemoteTaskRunnerWorkItem
>
asList
(
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
,
null
).
withQueueInsertionTime
(
new
DateTime
())
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
).
withQueueInsertionTime
(
new
DateTime
())
),
Arrays
.<
ZkWorker
>
asList
(
new
TestZkWorker
(
NoopTask
.
create
()),
...
...
@@ -412,7 +412,7 @@ public class SimpleResourceManagementStrategyTest
boolean
terminatedSomething
=
simpleResourceManagementStrategy
.
doTerminate
(
Arrays
.<
RemoteTaskRunnerWorkItem
>
asList
(
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
,
null
).
withQueueInsertionTime
(
new
DateTime
())
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
).
withQueueInsertionTime
(
new
DateTime
())
),
Arrays
.<
ZkWorker
>
asList
(
new
TestZkWorker
(
null
)
...
...
@@ -421,7 +421,7 @@ public class SimpleResourceManagementStrategyTest
boolean
provisionedSomething
=
simpleResourceManagementStrategy
.
doProvision
(
Arrays
.<
RemoteTaskRunnerWorkItem
>
asList
(
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
,
null
).
withQueueInsertionTime
(
new
DateTime
())
new
RemoteTaskRunnerWorkItem
(
testTask
.
getId
(),
null
).
withQueueInsertionTime
(
new
DateTime
())
),
Arrays
.<
ZkWorker
>
asList
(
new
TestZkWorker
(
null
)
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录