Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
Forever310
druid
提交
584ccac8
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,发现更多精彩内容 >>
提交
584ccac8
编写于
8月 01, 2013
作者:
F
fjy
浏览文件
操作
浏览文件
下载
差异文件
move scanning of workers and tasks into RTR start, simplify bootstrap, make tests better
上级
2f8351b3
fba2bc96
变更
18
展开全部
隐藏空白更改
内联
并排
Showing
18 changed file
with
1059 addition
and
392 deletion
+1059
-392
indexing-common/src/main/java/com/metamx/druid/index/QueryableIndex.java
.../src/main/java/com/metamx/druid/index/QueryableIndex.java
+10
-0
indexing-common/src/main/java/com/metamx/druid/index/SimpleQueryableIndex.java
...ain/java/com/metamx/druid/index/SimpleQueryableIndex.java
+12
-1
indexing-common/src/main/java/com/metamx/druid/index/v1/IndexIO.java
...mmon/src/main/java/com/metamx/druid/index/v1/IndexIO.java
+5
-3
indexing-common/src/main/java/com/metamx/druid/index/v1/MMappedIndex.java
...src/main/java/com/metamx/druid/index/v1/MMappedIndex.java
+20
-3
indexing-service/src/main/java/com/metamx/druid/indexing/coordinator/RemoteTaskRunner.java
...m/metamx/druid/indexing/coordinator/RemoteTaskRunner.java
+67
-62
indexing-service/src/main/java/com/metamx/druid/indexing/worker/WorkerCuratorCoordinator.java
...etamx/druid/indexing/worker/WorkerCuratorCoordinator.java
+1
-0
indexing-service/src/test/java/com/metamx/druid/indexing/coordinator/RemoteTaskRunnerTest.java
...tamx/druid/indexing/coordinator/RemoteTaskRunnerTest.java
+155
-204
pom.xml
pom.xml
+1
-1
server/src/main/java/com/metamx/druid/coordination/ServerManager.java
...ain/java/com/metamx/druid/coordination/ServerManager.java
+55
-28
server/src/main/java/com/metamx/druid/index/IncrementalIndexSegment.java
.../java/com/metamx/druid/index/IncrementalIndexSegment.java
+8
-0
server/src/main/java/com/metamx/druid/index/QueryableIndexSegment.java
...in/java/com/metamx/druid/index/QueryableIndexSegment.java
+9
-0
server/src/main/java/com/metamx/druid/index/ReferenceCountingSegment.java
...java/com/metamx/druid/index/ReferenceCountingSegment.java
+167
-0
server/src/main/java/com/metamx/druid/index/ReferenceCountingSequence.java
...ava/com/metamx/druid/index/ReferenceCountingSequence.java
+48
-0
server/src/main/java/com/metamx/druid/index/Segment.java
server/src/main/java/com/metamx/druid/index/Segment.java
+3
-1
server/src/main/java/com/metamx/druid/loading/QueryableIndexFactory.java
.../java/com/metamx/druid/loading/QueryableIndexFactory.java
+1
-0
server/src/test/java/com/metamx/druid/coordination/ServerManagerTest.java
...java/com/metamx/druid/coordination/ServerManagerTest.java
+354
-89
server/src/test/java/com/metamx/druid/index/ReferenceCountingSegmentTest.java
.../com/metamx/druid/index/ReferenceCountingSegmentTest.java
+137
-0
server/src/test/java/com/metamx/druid/loading/CacheTestSegmentLoader.java
...java/com/metamx/druid/loading/CacheTestSegmentLoader.java
+6
-0
未找到文件。
indexing-common/src/main/java/com/metamx/druid/index/QueryableIndex.java
浏览文件 @
584ccac8
...
...
@@ -23,6 +23,9 @@ import com.metamx.druid.index.column.ColumnSelector;
import
com.metamx.druid.kv.Indexed
;
import
org.joda.time.Interval
;
import
java.io.Closeable
;
import
java.io.IOException
;
/**
*/
public
interface
QueryableIndex
extends
ColumnSelector
...
...
@@ -31,4 +34,11 @@ public interface QueryableIndex extends ColumnSelector
public
int
getNumRows
();
public
Indexed
<
String
>
getColumnNames
();
public
Indexed
<
String
>
getAvailableDimensions
();
/**
* The close method shouldn't actually be here as this is nasty. We will adjust it in the future.
* @throws IOException
*/
@Deprecated
public
void
close
()
throws
IOException
;
}
indexing-common/src/main/java/com/metamx/druid/index/SimpleQueryableIndex.java
浏览文件 @
584ccac8
...
...
@@ -19,10 +19,12 @@
package
com.metamx.druid.index
;
import
com.metamx.common.io.smoosh.SmooshedFileMapper
;
import
com.metamx.druid.index.column.Column
;
import
com.metamx.druid.kv.Indexed
;
import
org.joda.time.Interval
;
import
java.io.IOException
;
import
java.util.Map
;
/**
...
...
@@ -34,13 +36,15 @@ public class SimpleQueryableIndex implements QueryableIndex
private
final
Indexed
<
String
>
availableDimensions
;
private
final
Column
timeColumn
;
private
final
Map
<
String
,
Column
>
otherColumns
;
private
final
SmooshedFileMapper
fileMapper
;
public
SimpleQueryableIndex
(
Interval
dataInterval
,
Indexed
<
String
>
columnNames
,
Indexed
<
String
>
dimNames
,
Column
timeColumn
,
Map
<
String
,
Column
>
otherColumns
Map
<
String
,
Column
>
otherColumns
,
SmooshedFileMapper
fileMapper
)
{
this
.
dataInterval
=
dataInterval
;
...
...
@@ -48,6 +52,7 @@ public class SimpleQueryableIndex implements QueryableIndex
this
.
availableDimensions
=
dimNames
;
this
.
timeColumn
=
timeColumn
;
this
.
otherColumns
=
otherColumns
;
this
.
fileMapper
=
fileMapper
;
}
@Override
...
...
@@ -85,4 +90,10 @@ public class SimpleQueryableIndex implements QueryableIndex
{
return
otherColumns
.
get
(
columnName
);
}
@Override
public
void
close
()
throws
IOException
{
fileMapper
.
close
();
}
}
indexing-common/src/main/java/com/metamx/druid/index/v1/IndexIO.java
浏览文件 @
584ccac8
...
...
@@ -375,7 +375,8 @@ public class IndexIO
dimValueLookups
,
dimColumns
,
invertedIndexed
,
spatialIndexed
spatialIndexed
,
smooshedFiles
);
log
.
debug
(
"Mapped v8 index[%s] in %,d millis"
,
inDir
,
System
.
currentTimeMillis
()
-
startTime
);
...
...
@@ -761,7 +762,8 @@ public class IndexIO
.
setType
(
ValueType
.
LONG
)
.
setGenericColumn
(
new
LongGenericColumnSupplier
(
index
.
timestamps
))
.
build
(),
columns
columns
,
index
.
getFileMapper
()
);
}
}
...
...
@@ -795,7 +797,7 @@ public class IndexIO
}
final
QueryableIndex
index
=
new
SimpleQueryableIndex
(
dataInterval
,
cols
,
dims
,
deserializeColumn
(
mapper
,
smooshedFiles
.
mapFile
(
"__time"
)),
columns
dataInterval
,
cols
,
dims
,
deserializeColumn
(
mapper
,
smooshedFiles
.
mapFile
(
"__time"
)),
columns
,
smooshedFiles
);
log
.
debug
(
"Mapped v9 index[%s] in %,d millis"
,
inDir
,
System
.
currentTimeMillis
()
-
startTime
);
...
...
indexing-common/src/main/java/com/metamx/druid/index/v1/MMappedIndex.java
浏览文件 @
584ccac8
...
...
@@ -24,18 +24,19 @@ import com.google.common.collect.Iterables;
import
com.google.common.collect.Maps
;
import
com.google.common.primitives.Ints
;
import
com.metamx.collections.spatial.ImmutableRTree
;
import
com.metamx.common.io.smoosh.SmooshedFileMapper
;
import
com.metamx.common.logger.Logger
;
import
com.metamx.druid.kv.ConciseCompressedIndexedInts
;
import
com.metamx.druid.kv.GenericIndexed
;
import
com.metamx.druid.kv.Indexed
;
import
com.metamx.druid.kv.IndexedList
;
import
com.metamx.druid.kv.IndexedLongs
;
import
com.metamx.druid.kv.IndexedRTree
;
import
com.metamx.druid.kv.VSizeIndexed
;
import
com.metamx.druid.kv.VSizeIndexedInts
;
import
it.uniroma3.mat.extendedset.intset.ImmutableConciseSet
;
import
org.joda.time.Interval
;
import
java.io.IOException
;
import
java.nio.ByteOrder
;
import
java.nio.LongBuffer
;
import
java.util.Arrays
;
...
...
@@ -57,6 +58,7 @@ public class MMappedIndex
final
Map
<
String
,
VSizeIndexed
>
dimColumns
;
final
Map
<
String
,
GenericIndexed
<
ImmutableConciseSet
>>
invertedIndexes
;
final
Map
<
String
,
ImmutableRTree
>
spatialIndexes
;
final
SmooshedFileMapper
fileMapper
;
private
final
Map
<
String
,
Integer
>
metricIndexes
=
Maps
.
newHashMap
();
...
...
@@ -69,7 +71,8 @@ public class MMappedIndex
Map
<
String
,
GenericIndexed
<
String
>>
dimValueLookups
,
Map
<
String
,
VSizeIndexed
>
dimColumns
,
Map
<
String
,
GenericIndexed
<
ImmutableConciseSet
>>
invertedIndexes
,
Map
<
String
,
ImmutableRTree
>
spatialIndexes
Map
<
String
,
ImmutableRTree
>
spatialIndexes
,
SmooshedFileMapper
fileMapper
)
{
this
.
availableDimensions
=
availableDimensions
;
...
...
@@ -81,6 +84,7 @@ public class MMappedIndex
this
.
dimColumns
=
dimColumns
;
this
.
invertedIndexes
=
invertedIndexes
;
this
.
spatialIndexes
=
spatialIndexes
;
this
.
fileMapper
=
fileMapper
;
for
(
int
i
=
0
;
i
<
availableMetrics
.
size
();
i
++)
{
metricIndexes
.
put
(
availableMetrics
.
get
(
i
),
i
);
...
...
@@ -169,6 +173,18 @@ public class MMappedIndex
return
(
retVal
==
null
)
?
emptySet
:
retVal
;
}
public
SmooshedFileMapper
getFileMapper
()
{
return
fileMapper
;
}
public
void
close
()
throws
IOException
{
if
(
fileMapper
!=
null
)
{
fileMapper
.
close
();
}
}
public
static
MMappedIndex
fromIndex
(
Index
index
)
{
log
.
info
(
"Converting timestamps"
);
...
...
@@ -273,7 +289,8 @@ public class MMappedIndex
dimValueLookups
,
dimColumns
,
invertedIndexes
,
spatialIndexes
spatialIndexes
,
null
);
}
}
indexing-service/src/main/java/com/metamx/druid/indexing/coordinator/RemoteTaskRunner.java
浏览文件 @
584ccac8
...
...
@@ -21,7 +21,6 @@ package com.metamx.druid.indexing.coordinator;
import
com.fasterxml.jackson.databind.ObjectMapper
;
import
com.google.common.base.Charsets
;
import
com.google.common.base.Function
;
import
com.google.common.base.Joiner
;
import
com.google.common.base.Optional
;
import
com.google.common.base.Preconditions
;
...
...
@@ -136,11 +135,55 @@ public class RemoteTaskRunner implements TaskRunner, TaskLogProvider
@LifecycleStart
public
void
start
()
{
if
(
started
)
{
return
;
}
try
{
if
(
started
)
{
return
;
}
// Add listener for creation/deletion of workers
workerPathCache
.
getListenable
().
addListener
(
new
PathChildrenCacheListener
()
{
@Override
public
void
childEvent
(
CuratorFramework
client
,
final
PathChildrenCacheEvent
event
)
throws
Exception
{
Worker
worker
;
switch
(
event
.
getType
())
{
case
CHILD_ADDED:
worker
=
jsonMapper
.
readValue
(
event
.
getData
().
getData
(),
Worker
.
class
);
addWorker
(
worker
,
PathChildrenCache
.
StartMode
.
NORMAL
);
break
;
case
CHILD_REMOVED:
worker
=
jsonMapper
.
readValue
(
event
.
getData
().
getData
(),
Worker
.
class
);
removeWorker
(
worker
);
break
;
default
:
break
;
}
}
}
);
workerPathCache
.
start
(
PathChildrenCache
.
StartMode
.
BUILD_INITIAL_CACHE
);
started
=
true
;
for
(
ChildData
childData
:
workerPathCache
.
getCurrentData
())
{
final
Worker
worker
=
jsonMapper
.
readValue
(
childData
.
getData
(),
Worker
.
class
);
addWorker
(
worker
,
PathChildrenCache
.
StartMode
.
BUILD_INITIAL_CACHE
);
}
started
=
true
;
}
catch
(
Exception
e
)
{
throw
Throwables
.
propagate
(
e
);
}
}
@LifecycleStop
...
...
@@ -203,68 +246,26 @@ public class RemoteTaskRunner implements TaskRunner, TaskLogProvider
public
void
bootstrap
(
List
<
Task
>
tasks
)
{
try
{
// Add listener for creation/deletion of workers
workerPathCache
.
getListenable
().
addListener
(
new
PathChildrenCacheListener
()
{
@Override
public
void
childEvent
(
CuratorFramework
client
,
final
PathChildrenCacheEvent
event
)
throws
Exception
{
Worker
worker
;
switch
(
event
.
getType
())
{
case
CHILD_ADDED:
worker
=
jsonMapper
.
readValue
(
event
.
getData
().
getData
(),
Worker
.
class
);
log
.
info
(
"Worker[%s] reportin' for duty!"
,
worker
.
getHost
());
addWorker
(
worker
,
PathChildrenCache
.
StartMode
.
NORMAL
);
break
;
case
CHILD_REMOVED:
worker
=
jsonMapper
.
readValue
(
event
.
getData
().
getData
(),
Worker
.
class
);
log
.
info
(
"Kaboom! Worker[%s] removed!"
,
worker
.
getHost
());
removeWorker
(
worker
);
break
;
default
:
break
;
}
}
}
);
workerPathCache
.
start
(
PathChildrenCache
.
StartMode
.
BUILD_INITIAL_CACHE
);
if
(!
started
)
{
throw
new
ISE
(
"Must start RTR first before calling bootstrap!"
);
}
Set
<
String
>
existingTasks
=
Sets
.
newHashSet
();
for
(
ChildData
childData
:
workerPathCache
.
getCurrentData
())
{
final
Worker
worker
=
jsonMapper
.
readValue
(
childData
.
getData
(),
Worker
.
class
);
log
.
info
(
"Worker[%s] reportin' for duty!"
,
worker
.
getHost
());
final
ZkWorker
zkWorker
=
addWorker
(
worker
,
PathChildrenCache
.
StartMode
.
BUILD_INITIAL_CACHE
);
for
(
ZkWorker
zkWorker
:
zkWorkers
.
values
())
{
existingTasks
.
addAll
(
zkWorker
.
getRunningTasks
().
keySet
());
}
Set
<
String
>
bootstrappedTasks
=
Sets
.
newHashSet
(
Lists
.
transform
(
tasks
,
new
Function
<
Task
,
String
>()
{
@Override
public
String
apply
(
Task
input
)
{
return
input
.
getId
();
}
}
)
);
// shutdown any tasks that we don't know about
for
(
String
taskId
:
Sets
.
difference
(
existingTasks
,
bootstrappedTasks
))
{
shutdown
(
taskId
);
for
(
Task
task
:
tasks
)
{
if
(
existingTasks
.
contains
(
task
.
getId
()))
{
log
.
info
(
"Bootstrap found %s running."
,
task
.
getId
());
runningTasks
.
put
(
task
.
getId
(),
new
RemoteTaskRunnerWorkItem
(
task
,
SettableFuture
.<
TaskStatus
>
create
())
);
}
else
{
log
.
info
(
"Bootstrap didn't find %s running. Running it again"
,
task
.
getId
());
run
(
task
);
}
}
}
catch
(
Exception
e
)
{
...
...
@@ -530,6 +531,8 @@ public class RemoteTaskRunner implements TaskRunner, TaskLogProvider
*/
private
ZkWorker
addWorker
(
final
Worker
worker
,
PathChildrenCache
.
StartMode
startMode
)
{
log
.
info
(
"Worker[%s] reportin' for duty!"
,
worker
.
getHost
());
try
{
final
String
workerStatusPath
=
JOINER
.
join
(
config
.
getIndexerStatusPath
(),
worker
.
getHost
());
final
PathChildrenCache
statusCache
=
pathChildrenCacheFactory
.
make
(
cf
,
workerStatusPath
);
...
...
@@ -631,6 +634,8 @@ public class RemoteTaskRunner implements TaskRunner, TaskLogProvider
*/
private
void
removeWorker
(
final
Worker
worker
)
{
log
.
info
(
"Kaboom! Worker[%s] removed!"
,
worker
.
getHost
());
ZkWorker
zkWorker
=
zkWorkers
.
get
(
worker
.
getHost
());
if
(
zkWorker
!=
null
)
{
try
{
...
...
indexing-service/src/main/java/com/metamx/druid/indexing/worker/WorkerCuratorCoordinator.java
浏览文件 @
584ccac8
...
...
@@ -130,6 +130,7 @@ public class WorkerCuratorCoordinator
}
curatorFramework
.
create
()
.
creatingParentsIfNeeded
()
.
withMode
(
mode
)
.
forPath
(
path
,
rawBytes
);
}
...
...
indexing-service/src/test/java/com/metamx/druid/indexing/coordinator/RemoteTaskRunnerTest.java
浏览文件 @
584ccac8
...
...
@@ -2,24 +2,23 @@ package com.metamx.druid.indexing.coordinator;
import
com.fasterxml.jackson.databind.ObjectMapper
;
import
com.fasterxml.jackson.databind.jsontype.NamedType
;
import
com.google.common.base.Function
;
import
com.google.common.base.Joiner
;
import
com.google.common.base.Stopwatch
;
import
com.google.common.
base.Throw
ables
;
import
com.google.common.
collect.Iter
ables
;
import
com.google.common.collect.Lists
;
import
com.google.common.collect.Sets
;
import
com.google.common.io.Files
;
import
com.google.common.util.concurrent.FutureCallback
;
import
com.google.common.util.concurrent.Futures
;
import
com.google.common.util.concurrent.ListenableFuture
;
import
com.metamx.common.ISE
;
import
com.metamx.druid.aggregation.AggregatorFactory
;
import
com.metamx.druid.client.DataSegment
;
import
com.metamx.druid.curator.PotentiallyGzippedCompressionProvider
;
import
com.metamx.druid.curator.cache.PathChildrenCacheFactory
;
import
com.metamx.druid.curator.cache.SimplePathChildrenCacheFactory
;
import
com.metamx.druid.indexing.TestTask
;
import
com.metamx.druid.indexing.common.RetryPolicyFactory
;
import
com.metamx.druid.indexing.common.TaskStatus
;
import
com.metamx.druid.indexing.common.TaskToolboxFactory
;
import
com.metamx.druid.indexing.common.config.IndexerZkConfig
;
import
com.metamx.druid.indexing.common.config.RetryPolicyConfig
;
import
com.metamx.druid.indexing.common.config.TaskConfig
;
import
com.metamx.druid.indexing.common.task.Task
;
import
com.metamx.druid.indexing.common.task.TaskResource
;
...
...
@@ -31,7 +30,6 @@ import com.metamx.druid.indexing.worker.WorkerTaskMonitor;
import
com.metamx.druid.jackson.DefaultObjectMapper
;
import
com.metamx.emitter.EmittingLogger
;
import
com.metamx.emitter.service.ServiceEmitter
;
import
org.apache.commons.lang.mutable.MutableBoolean
;
import
org.apache.curator.framework.CuratorFramework
;
import
org.apache.curator.framework.CuratorFrameworkFactory
;
import
org.apache.curator.framework.recipes.cache.PathChildrenCache
;
...
...
@@ -47,33 +45,34 @@ import org.junit.Assert;
import
org.junit.Before
;
import
org.junit.Test
;
import
javax.annotation.Nullable
;
import
java.io.File
;
import
java.util.Arrays
;
import
java.util.Set
;
import
java.util.concurrent.Executors
;
import
java.util.concurrent.ScheduledExecutorService
;
import
java.util.concurrent.TimeUnit
;
import
java.util.concurrent.atomic.AtomicReference
;
import
static
junit
.
framework
.
Assert
.
fail
;
/**
*/
public
class
RemoteTaskRunnerTest
{
private
static
final
ObjectMapper
jsonMapper
=
new
DefaultObjectMapper
();
private
static
final
String
basePath
=
"/test/druid/indexer"
;
private
static
final
String
announcementsPath
=
String
.
format
(
"%s/announcements"
,
basePath
);
private
static
final
String
tasksPath
=
String
.
format
(
"%s/tasks"
,
basePath
);
private
static
final
String
statusPath
=
String
.
format
(
"%s/status"
,
basePath
);
private
static
final
Joiner
joiner
=
Joiner
.
on
(
"/"
);
private
static
final
String
basePath
=
"/test/druid"
;
private
static
final
String
announcementsPath
=
String
.
format
(
"%s/indexer/announcements/worker"
,
basePath
);
private
static
final
String
tasksPath
=
String
.
format
(
"%s/indexer/tasks/worker"
,
basePath
);
private
static
final
String
statusPath
=
String
.
format
(
"%s/indexer/status/worker"
,
basePath
);
private
TestingCluster
testingCluster
;
private
CuratorFramework
cf
;
private
PathChildrenCacheFactory
pathChildrenCacheFactory
;
private
RemoteTaskRunner
remoteTaskRunner
;
private
WorkerCuratorCoordinator
workerCuratorCoordinator
;
private
WorkerTaskMonitor
workerTaskMonitor
;
private
TestTask
task
1
;
private
TestTask
task
;
private
Worker
worker
1
;
private
Worker
worker
;
@Before
public
void
setUp
()
throws
Exception
...
...
@@ -87,78 +86,38 @@ public class RemoteTaskRunnerTest
.
compressionProvider
(
new
PotentiallyGzippedCompressionProvider
(
false
))
.
build
();
cf
.
start
();
cf
.
create
().
creatingParentsIfNeeded
().
forPath
(
basePath
);
cf
.
create
().
creatingParentsIfNeeded
().
forPath
(
announcementsPath
);
cf
.
create
().
forPath
(
tasksPath
);
cf
.
create
().
forPath
(
String
.
format
(
"%s/worker1"
,
tasksPath
));
cf
.
create
().
forPath
(
statusPath
);
cf
.
create
().
forPath
(
String
.
format
(
"%s/worker1"
,
statusPath
));
pathChildrenCacheFactory
=
new
SimplePathChildrenCacheFactory
.
Builder
().
build
();
worker1
=
new
Worker
(
"worker1"
,
"localhost"
,
3
,
"0"
);
task1
=
new
TestTask
(
"task1"
,
"dummyDs"
,
Lists
.<
DataSegment
>
newArrayList
(
new
DataSegment
(
"dummyDs"
,
new
Interval
(
new
DateTime
(),
new
DateTime
()),
new
DateTime
().
toString
(),
null
,
null
,
null
,
null
,
0
,
0
)
),
Lists
.<
AggregatorFactory
>
newArrayList
(),
TaskStatus
.
success
(
"task1"
)
);
makeRemoteTaskRunner
();
makeTaskMonitor
();
task
=
makeTask
(
TaskStatus
.
success
(
"task"
));
}
@After
public
void
tearDown
()
throws
Exception
{
testingCluster
.
stop
();
remoteTaskRunner
.
stop
();
workerCuratorCoordinator
.
stop
();
workerTaskMonitor
.
stop
();
cf
.
close
();
testingCluster
.
stop
();
}
@Test
public
void
testRunNoExistingTask
()
throws
Exception
{
remoteTaskRunner
.
run
(
task1
);
doSetup
();
remoteTaskRunner
.
run
(
task
);
}
@Test
@Test
(
expected
=
ISE
.
class
)
public
void
testExceptionThrownWithExistingTask
()
throws
Exception
{
doSetup
();
remoteTaskRunner
.
run
(
makeTask
(
TaskStatus
.
running
(
"task"
)));
remoteTaskRunner
.
run
(
new
TestTask
(
task1
.
getId
(),
task1
.
getDataSource
(),
task1
.
getSegments
(),
Lists
.<
AggregatorFactory
>
newArrayList
(),
TaskStatus
.
running
(
task1
.
getId
())
)
makeTask
(
TaskStatus
.
running
(
"task"
))
);
try
{
remoteTaskRunner
.
run
(
task1
);
fail
(
"ISE expected"
);
}
catch
(
ISE
expected
)
{
}
}
@Test
...
...
@@ -167,94 +126,19 @@ public class RemoteTaskRunnerTest
ServiceEmitter
emitter
=
EasyMock
.
createMock
(
ServiceEmitter
.
class
);
EmittingLogger
.
registerEmitter
(
emitter
);
EasyMock
.
replay
(
emitter
);
remoteTaskRunner
.
run
(
new
TestTask
(
new
String
(
new
char
[
5000
]),
"dummyDs"
,
Lists
.<
DataSegment
>
newArrayList
(
new
DataSegment
(
"dummyDs"
,
new
Interval
(
new
DateTime
(),
new
DateTime
()),
new
DateTime
().
toString
(),
null
,
null
,
null
,
null
,
0
,
0
)
),
Lists
.<
AggregatorFactory
>
newArrayList
(),
TaskStatus
.
success
(
"foo"
)
)
);
EasyMock
.
verify
(
emitter
);
}
@Test
public
void
testRunWithCallback
()
throws
Exception
{
final
MutableBoolean
callbackCalled
=
new
MutableBoolean
(
false
);
Futures
.
addCallback
(
remoteTaskRunner
.
run
(
new
TestTask
(
task1
.
getId
(),
task1
.
getDataSource
(),
task1
.
getSegments
(),
Lists
.<
AggregatorFactory
>
newArrayList
(),
TaskStatus
.
running
(
task1
.
getId
())
)
),
new
FutureCallback
<
TaskStatus
>()
{
@Override
public
void
onSuccess
(
TaskStatus
taskStatus
)
{
callbackCalled
.
setValue
(
true
);
}
@Override
public
void
onFailure
(
Throwable
throwable
)
{
// neg
}
}
);
// Really don't like this way of waiting for the task to appear
int
count
=
0
;
while
(
remoteTaskRunner
.
findWorkerRunningTask
(
task1
.
getId
())
==
null
)
{
Thread
.
sleep
(
500
);
if
(
count
>
10
)
{
throw
new
ISE
(
"WTF?! Task still not announced in ZK?"
);
}
count
++;
}
Assert
.
assertTrue
(
remoteTaskRunner
.
getRunningTasks
().
size
()
==
1
);
// Complete the task
cf
.
setData
().
forPath
(
String
.
format
(
"%s/worker1/task1"
,
statusPath
),
jsonMapper
.
writeValueAsBytes
(
TaskStatus
.
success
(
task1
.
getId
()))
);
doSetup
();
// Really don't like this way of waiting for the task to disappear
count
=
0
;
while
(
remoteTaskRunner
.
findWorkerRunningTask
(
task1
.
getId
())
!=
null
)
{
Thread
.
sleep
(
500
);
if
(
count
>
10
)
{
throw
new
ISE
(
"WTF?! Task still exists in ZK?"
);
}
count
++;
}
remoteTaskRunner
.
run
(
makeTask
(
TaskStatus
.
success
(
new
String
(
new
char
[
5000
]))));
Assert
.
assertTrue
(
"TaskCallback was not called!"
,
callbackCalled
.
booleanValue
()
);
EasyMock
.
verify
(
emitter
);
}
@Test
public
void
testRunSameAvailabilityGroup
()
throws
Exception
{
doSetup
();
TestRealtimeTask
theTask
=
new
TestRealtimeTask
(
"rt1"
,
new
TaskResource
(
"rt1"
,
1
),
...
...
@@ -286,6 +170,8 @@ public class RemoteTaskRunnerTest
@Test
public
void
testRunWithCapacity
()
throws
Exception
{
doSetup
();
TestRealtimeTask
theTask
=
new
TestRealtimeTask
(
"rt1"
,
new
TaskResource
(
"rt1"
,
1
),
...
...
@@ -314,34 +200,115 @@ public class RemoteTaskRunnerTest
Assert
.
assertTrue
(
remoteTaskRunner
.
getPendingTasks
().
iterator
().
next
().
getTask
().
getId
().
equals
(
"rt2"
));
}
@Test
public
void
testFailure
()
throws
Exception
{
doSetup
();
ListenableFuture
<
TaskStatus
>
future
=
remoteTaskRunner
.
run
(
makeTask
(
TaskStatus
.
running
(
"task"
)));
final
String
taskStatus
=
joiner
.
join
(
statusPath
,
"task"
);
Stopwatch
stopwatch
=
new
Stopwatch
();
stopwatch
.
start
();
while
(
cf
.
checkExists
().
forPath
(
taskStatus
)
==
null
)
{
Thread
.
sleep
(
100
);
if
(
stopwatch
.
elapsed
(
TimeUnit
.
MILLISECONDS
)
>
1000
)
{
throw
new
ISE
(
"Cannot find running task"
);
}
}
Assert
.
assertTrue
(
remoteTaskRunner
.
getRunningTasks
().
iterator
().
next
().
getTask
().
getId
().
equals
(
"task"
));
cf
.
delete
().
forPath
(
taskStatus
);
TaskStatus
status
=
future
.
get
();
Assert
.
assertEquals
(
status
.
getStatusCode
(),
TaskStatus
.
Status
.
FAILED
);
}
@Test
public
void
testBootstrap
()
throws
Exception
{
cf
.
create
()
.
creatingParentsIfNeeded
()
.
withMode
(
CreateMode
.
EPHEMERAL
)
.
forPath
(
joiner
.
join
(
statusPath
,
"first"
),
jsonMapper
.
writeValueAsBytes
(
TaskStatus
.
running
(
"first"
)));
cf
.
create
()
.
creatingParentsIfNeeded
()
.
withMode
(
CreateMode
.
EPHEMERAL
)
.
forPath
(
joiner
.
join
(
statusPath
,
"second"
),
jsonMapper
.
writeValueAsBytes
(
TaskStatus
.
running
(
"second"
)));
doSetup
();
Set
<
String
>
existingTasks
=
Sets
.
newHashSet
();
for
(
ZkWorker
zkWorker
:
remoteTaskRunner
.
getWorkers
())
{
existingTasks
.
addAll
(
zkWorker
.
getRunningTasks
().
keySet
());
}
Assert
.
assertTrue
(
existingTasks
.
size
()
==
2
);
Assert
.
assertTrue
(
existingTasks
.
contains
(
"first"
));
Assert
.
assertTrue
(
existingTasks
.
contains
(
"second"
));
remoteTaskRunner
.
bootstrap
(
Arrays
.<
Task
>
asList
(
makeTask
(
TaskStatus
.
running
(
"second"
))));
Set
<
String
>
runningTasks
=
Sets
.
newHashSet
(
Iterables
.
transform
(
remoteTaskRunner
.
getRunningTasks
(),
new
Function
<
RemoteTaskRunnerWorkItem
,
String
>()
{
@Override
public
String
apply
(
RemoteTaskRunnerWorkItem
input
)
{
return
input
.
getTask
().
getId
();
}
}
)
);
Assert
.
assertTrue
(
runningTasks
.
size
()
==
1
);
Assert
.
assertTrue
(
runningTasks
.
contains
(
"second"
));
Assert
.
assertFalse
(
runningTasks
.
contains
(
"first"
));
}
private
void
doSetup
()
throws
Exception
{
makeWorker
();
makeRemoteTaskRunner
();
makeTaskMonitor
();
}
private
TestTask
makeTask
(
TaskStatus
status
)
{
return
new
TestTask
(
status
.
getId
(),
"dummyDs"
,
Lists
.<
DataSegment
>
newArrayList
(
new
DataSegment
(
"dummyDs"
,
new
Interval
(
new
DateTime
(),
new
DateTime
()),
new
DateTime
().
toString
(),
null
,
null
,
null
,
null
,
0
,
0
)
),
Lists
.<
AggregatorFactory
>
newArrayList
(),
status
);
}
private
void
makeTaskMonitor
()
throws
Exception
{
WorkerCuratorCoordinator
workerCuratorCoordinator
=
new
WorkerCuratorCoordinator
(
workerCuratorCoordinator
=
new
WorkerCuratorCoordinator
(
jsonMapper
,
new
IndexerZkConfig
()
{
@Override
public
String
getIndexerAnnouncementPath
()
{
return
announcementsPath
;
}
@Override
public
String
getIndexerTaskPath
()
{
return
tasksPath
;
}
@Override
public
String
getIndexerStatusPath
()
{
return
statusPath
;
}
@Override
public
String
getZkBasePath
()
{
throw
new
UnsupportedOperationException
()
;
return
basePath
;
}
@Override
...
...
@@ -351,13 +318,14 @@ public class RemoteTaskRunnerTest
}
},
cf
,
worker
1
worker
);
workerCuratorCoordinator
.
start
();
// Start a task monitor
workerTaskMonitor
=
new
WorkerTaskMonitor
(
jsonMapper
,
new
PathChildrenCache
(
cf
,
String
.
format
(
"%s/worker1"
,
tasksPath
)
,
true
),
new
PathChildrenCache
(
cf
,
tasksPath
,
true
),
cf
,
workerCuratorCoordinator
,
new
ThreadPoolTaskRunner
(
...
...
@@ -399,26 +367,27 @@ public class RemoteTaskRunnerTest
jsonMapper
,
new
TestRemoteTaskRunnerConfig
(),
cf
,
pathChildrenCacheFactory
,
new
SimplePathChildrenCacheFactory
.
Builder
().
build
()
,
new
AtomicReference
<
WorkerSetupData
>(
new
WorkerSetupData
(
"0"
,
0
,
1
,
null
,
null
)),
null
);
// Create a single worker and wait for things for be ready
remoteTaskRunner
.
start
();
remoteTaskRunner
.
bootstrap
(
Lists
.<
Task
>
newArrayList
());
}
private
void
makeWorker
()
throws
Exception
{
worker
=
new
Worker
(
"worker"
,
"localhost"
,
3
,
"0"
);
cf
.
create
().
creatingParentsIfNeeded
().
withMode
(
CreateMode
.
EPHEMERAL
).
forPath
(
String
.
format
(
"%s/worker1"
,
announcementsPath
)
,
jsonMapper
.
writeValueAsBytes
(
worker
1
)
announcementsPath
,
jsonMapper
.
writeValueAsBytes
(
worker
)
);
int
count
=
0
;
while
(
remoteTaskRunner
.
getWorkers
().
size
()
==
0
)
{
Thread
.
sleep
(
500
);
if
(
count
>
10
)
{
throw
new
ISE
(
"WTF?! Still can't find worker!"
);
}
count
++;
}
}
private
static
class
TestRemoteTaskRunnerConfig
extends
RemoteTaskRunnerConfig
...
...
@@ -429,28 +398,10 @@ public class RemoteTaskRunnerTest
return
false
;
}
@Override
public
String
getIndexerAnnouncementPath
()
{
return
announcementsPath
;
}
@Override
public
String
getIndexerTaskPath
()
{
return
tasksPath
;
}
@Override
public
String
getIndexerStatusPath
()
{
return
statusPath
;
}
@Override
public
String
getZkBasePath
()
{
throw
new
UnsupportedOperationException
()
;
return
basePath
;
}
@Override
...
...
pom.xml
浏览文件 @
584ccac8
...
...
@@ -38,7 +38,7 @@
<properties>
<project.build.sourceEncoding>
UTF-8
</project.build.sourceEncoding>
<metamx.java-util.version>
0.22.
4
</metamx.java-util.version>
<metamx.java-util.version>
0.22.
6
</metamx.java-util.version>
<apache.curator.version>
2.1.0-incubating
</apache.curator.version>
</properties>
...
...
server/src/main/java/com/metamx/druid/coordination/ServerManager.java
浏览文件 @
584ccac8
...
...
@@ -24,11 +24,14 @@ import com.google.common.base.Predicates;
import
com.google.common.collect.Ordering
;
import
com.metamx.common.ISE
;
import
com.metamx.common.guava.FunctionalIterable
;
import
com.metamx.common.guava.Sequence
;
import
com.metamx.druid.Query
;
import
com.metamx.druid.TimelineObjectHolder
;
import
com.metamx.druid.VersionedIntervalTimeline
;
import
com.metamx.druid.client.DataSegment
;
import
com.metamx.druid.collect.CountingMap
;
import
com.metamx.druid.index.ReferenceCountingSegment
;
import
com.metamx.druid.index.ReferenceCountingSequence
;
import
com.metamx.druid.index.Segment
;
import
com.metamx.druid.loading.SegmentLoader
;
import
com.metamx.druid.loading.SegmentLoadingException
;
...
...
@@ -53,6 +56,7 @@ import com.metamx.emitter.service.ServiceMetricEvent;
import
org.joda.time.Interval
;
import
javax.annotation.Nullable
;
import
java.io.IOException
;
import
java.util.Arrays
;
import
java.util.HashMap
;
import
java.util.Map
;
...
...
@@ -71,7 +75,7 @@ public class ServerManager implements QuerySegmentWalker
private
final
ServiceEmitter
emitter
;
private
final
ExecutorService
exec
;
private
final
Map
<
String
,
VersionedIntervalTimeline
<
String
,
Segment
>>
dataSources
;
private
final
Map
<
String
,
VersionedIntervalTimeline
<
String
,
ReferenceCounting
Segment
>>
dataSources
;
private
final
CountingMap
<
String
>
dataSourceSizes
=
new
CountingMap
<
String
>();
private
final
CountingMap
<
String
>
dataSourceCounts
=
new
CountingMap
<
String
>();
...
...
@@ -88,7 +92,7 @@ public class ServerManager implements QuerySegmentWalker
this
.
exec
=
exec
;
this
.
dataSources
=
new
HashMap
<
String
,
VersionedIntervalTimeline
<
String
,
Segment
>>();
this
.
dataSources
=
new
HashMap
<
String
,
VersionedIntervalTimeline
<
String
,
ReferenceCounting
Segment
>>();
}
public
Map
<
String
,
Long
>
getDataSourceSizes
()
...
...
@@ -132,14 +136,14 @@ public class ServerManager implements QuerySegmentWalker
synchronized
(
lock
)
{
String
dataSource
=
segment
.
getDataSource
();
VersionedIntervalTimeline
<
String
,
Segment
>
loadedIntervals
=
dataSources
.
get
(
dataSource
);
VersionedIntervalTimeline
<
String
,
ReferenceCounting
Segment
>
loadedIntervals
=
dataSources
.
get
(
dataSource
);
if
(
loadedIntervals
==
null
)
{
loadedIntervals
=
new
VersionedIntervalTimeline
<
String
,
Segment
>(
Ordering
.
natural
());
loadedIntervals
=
new
VersionedIntervalTimeline
<
String
,
ReferenceCounting
Segment
>(
Ordering
.
natural
());
dataSources
.
put
(
dataSource
,
loadedIntervals
);
}
PartitionHolder
<
Segment
>
entry
=
loadedIntervals
.
findEntry
(
PartitionHolder
<
ReferenceCounting
Segment
>
entry
=
loadedIntervals
.
findEntry
(
segment
.
getInterval
(),
segment
.
getVersion
()
);
...
...
@@ -149,7 +153,9 @@ public class ServerManager implements QuerySegmentWalker
}
loadedIntervals
.
add
(
segment
.
getInterval
(),
segment
.
getVersion
(),
segment
.
getShardSpec
().
createChunk
(
adapter
)
segment
.
getInterval
(),
segment
.
getVersion
(),
segment
.
getShardSpec
().
createChunk
(
new
ReferenceCountingSegment
(
adapter
))
);
synchronized
(
dataSourceSizes
)
{
dataSourceSizes
.
add
(
dataSource
,
segment
.
getSize
());
...
...
@@ -164,17 +170,19 @@ public class ServerManager implements QuerySegmentWalker
{
String
dataSource
=
segment
.
getDataSource
();
synchronized
(
lock
)
{
VersionedIntervalTimeline
<
String
,
Segment
>
loadedIntervals
=
dataSources
.
get
(
dataSource
);
VersionedIntervalTimeline
<
String
,
ReferenceCounting
Segment
>
loadedIntervals
=
dataSources
.
get
(
dataSource
);
if
(
loadedIntervals
==
null
)
{
log
.
info
(
"Told to delete a queryable for a dataSource[%s] that doesn't exist."
,
dataSource
);
return
;
}
PartitionChunk
<
Segment
>
removed
=
loadedIntervals
.
remove
(
segment
.
getInterval
(),
segment
.
getVersion
(),
segment
.
getShardSpec
().
createChunk
((
Segment
)
null
)
PartitionChunk
<
ReferenceCountingSegment
>
removed
=
loadedIntervals
.
remove
(
segment
.
getInterval
(),
segment
.
getVersion
(),
segment
.
getShardSpec
().
createChunk
((
ReferenceCountingSegment
)
null
)
);
Segment
oldQueryable
=
(
removed
==
null
)
?
null
:
removed
.
getObject
();
ReferenceCounting
Segment
oldQueryable
=
(
removed
==
null
)
?
null
:
removed
.
getObject
();
if
(
oldQueryable
!=
null
)
{
synchronized
(
dataSourceSizes
)
{
...
...
@@ -183,6 +191,16 @@ public class ServerManager implements QuerySegmentWalker
synchronized
(
dataSourceCounts
)
{
dataSourceCounts
.
add
(
dataSource
,
-
1L
);
}
try
{
oldQueryable
.
close
();
}
catch
(
IOException
e
)
{
log
.
makeAlert
(
e
,
"Exception closing segment"
)
.
addData
(
"dataSource"
,
dataSource
)
.
addData
(
"segmentId"
,
segment
.
getIdentifier
())
.
emit
();
}
}
else
{
log
.
info
(
"Told to delete a queryable on dataSource[%s] for interval[%s] and version [%s] that I don't have."
,
...
...
@@ -205,7 +223,7 @@ public class ServerManager implements QuerySegmentWalker
final
QueryToolChest
<
T
,
Query
<
T
>>
toolChest
=
factory
.
getToolchest
();
final
VersionedIntervalTimeline
<
String
,
Segment
>
timeline
=
dataSources
.
get
(
query
.
getDataSource
());
final
VersionedIntervalTimeline
<
String
,
ReferenceCounting
Segment
>
timeline
=
dataSources
.
get
(
query
.
getDataSource
());
if
(
timeline
==
null
)
{
return
new
NoopQueryRunner
<
T
>();
...
...
@@ -214,20 +232,22 @@ public class ServerManager implements QuerySegmentWalker
FunctionalIterable
<
QueryRunner
<
T
>>
adapters
=
FunctionalIterable
.
create
(
intervals
)
.
transformCat
(
new
Function
<
Interval
,
Iterable
<
TimelineObjectHolder
<
String
,
Segment
>>>()
new
Function
<
Interval
,
Iterable
<
TimelineObjectHolder
<
String
,
ReferenceCounting
Segment
>>>()
{
@Override
public
Iterable
<
TimelineObjectHolder
<
String
,
Segment
>>
apply
(
Interval
input
)
public
Iterable
<
TimelineObjectHolder
<
String
,
ReferenceCounting
Segment
>>
apply
(
Interval
input
)
{
return
timeline
.
lookup
(
input
);
}
}
)
.
transformCat
(
new
Function
<
TimelineObjectHolder
<
String
,
Segment
>,
Iterable
<
QueryRunner
<
T
>>>()
new
Function
<
TimelineObjectHolder
<
String
,
ReferenceCounting
Segment
>,
Iterable
<
QueryRunner
<
T
>>>()
{
@Override
public
Iterable
<
QueryRunner
<
T
>>
apply
(
@Nullable
final
TimelineObjectHolder
<
String
,
Segment
>
holder
)
public
Iterable
<
QueryRunner
<
T
>>
apply
(
@Nullable
final
TimelineObjectHolder
<
String
,
ReferenceCountingSegment
>
holder
)
{
if
(
holder
==
null
)
{
return
null
;
...
...
@@ -236,10 +256,10 @@ public class ServerManager implements QuerySegmentWalker
return
FunctionalIterable
.
create
(
holder
.
getObject
())
.
transform
(
new
Function
<
PartitionChunk
<
Segment
>,
QueryRunner
<
T
>>()
new
Function
<
PartitionChunk
<
ReferenceCounting
Segment
>,
QueryRunner
<
T
>>()
{
@Override
public
QueryRunner
<
T
>
apply
(
PartitionChunk
<
Segment
>
input
)
public
QueryRunner
<
T
>
apply
(
PartitionChunk
<
ReferenceCounting
Segment
>
input
)
{
return
buildAndDecorateQueryRunner
(
factory
,
...
...
@@ -280,7 +300,7 @@ public class ServerManager implements QuerySegmentWalker
final
QueryToolChest
<
T
,
Query
<
T
>>
toolChest
=
factory
.
getToolchest
();
final
VersionedIntervalTimeline
<
String
,
Segment
>
timeline
=
dataSources
.
get
(
query
.
getDataSource
());
final
VersionedIntervalTimeline
<
String
,
ReferenceCounting
Segment
>
timeline
=
dataSources
.
get
(
query
.
getDataSource
());
if
(
timeline
==
null
)
{
return
new
NoopQueryRunner
<
T
>();
...
...
@@ -293,9 +313,9 @@ public class ServerManager implements QuerySegmentWalker
{
@Override
@SuppressWarnings
(
"unchecked"
)
public
Iterable
<
QueryRunner
<
T
>>
apply
(
@Nullable
SegmentDescriptor
input
)
public
Iterable
<
QueryRunner
<
T
>>
apply
(
SegmentDescriptor
input
)
{
final
PartitionHolder
<
Segment
>
entry
=
timeline
.
findEntry
(
final
PartitionHolder
<
ReferenceCounting
Segment
>
entry
=
timeline
.
findEntry
(
input
.
getInterval
(),
input
.
getVersion
()
);
...
...
@@ -303,12 +323,12 @@ public class ServerManager implements QuerySegmentWalker
return
null
;
}
final
PartitionChunk
<
Segment
>
chunk
=
entry
.
getChunk
(
input
.
getPartitionNumber
());
final
PartitionChunk
<
ReferenceCounting
Segment
>
chunk
=
entry
.
getChunk
(
input
.
getPartitionNumber
());
if
(
chunk
==
null
)
{
return
null
;
}
final
Segment
adapter
=
chunk
.
getObject
();
final
ReferenceCounting
Segment
adapter
=
chunk
.
getObject
();
return
Arrays
.
asList
(
buildAndDecorateQueryRunner
(
factory
,
toolChest
,
adapter
,
new
SpecificSegmentSpec
(
input
))
);
...
...
@@ -323,10 +343,10 @@ public class ServerManager implements QuerySegmentWalker
}
private
<
T
>
QueryRunner
<
T
>
buildAndDecorateQueryRunner
(
QueryRunnerFactory
<
T
,
Query
<
T
>>
factory
,
final
QueryRunnerFactory
<
T
,
Query
<
T
>>
factory
,
final
QueryToolChest
<
T
,
Query
<
T
>>
toolChest
,
Segment
adapter
,
QuerySegmentSpec
segmentSpec
final
ReferenceCounting
Segment
adapter
,
final
QuerySegmentSpec
segmentSpec
)
{
return
new
SpecificSegmentQueryRunner
<
T
>(
...
...
@@ -335,7 +355,7 @@ public class ServerManager implements QuerySegmentWalker
new
Function
<
Query
<
T
>,
ServiceMetricEvent
.
Builder
>()
{
@Override
public
ServiceMetricEvent
.
Builder
apply
(
@Nullable
Query
<
T
>
input
)
public
ServiceMetricEvent
.
Builder
apply
(
@Nullable
final
Query
<
T
>
input
)
{
return
toolChest
.
makeMetricBuilder
(
input
);
}
...
...
@@ -343,10 +363,17 @@ public class ServerManager implements QuerySegmentWalker
new
BySegmentQueryRunner
<
T
>(
adapter
.
getIdentifier
(),
adapter
.
getDataInterval
().
getStart
(),
factory
.
createRunner
(
adapter
)
new
QueryRunner
<
T
>()
{
@Override
public
Sequence
<
T
>
run
(
final
Query
<
T
>
query
)
{
return
new
ReferenceCountingSequence
<
T
>(
factory
.
createRunner
(
adapter
).
run
(
query
),
adapter
);
}
}
)
).
withWaitMeasuredFromNow
(),
segmentSpec
);
}
}
}
\ No newline at end of file
server/src/main/java/com/metamx/druid/index/IncrementalIndexSegment.java
浏览文件 @
584ccac8
...
...
@@ -24,6 +24,8 @@ import com.metamx.druid.index.v1.IncrementalIndex;
import
com.metamx.druid.index.v1.IncrementalIndexStorageAdapter
;
import
org.joda.time.Interval
;
import
java.io.IOException
;
/**
*/
public
class
IncrementalIndexSegment
implements
Segment
...
...
@@ -60,4 +62,10 @@ public class IncrementalIndexSegment implements Segment
{
return
new
IncrementalIndexStorageAdapter
(
index
);
}
@Override
public
void
close
()
throws
IOException
{
// do nothing
}
}
server/src/main/java/com/metamx/druid/index/QueryableIndexSegment.java
浏览文件 @
584ccac8
...
...
@@ -23,6 +23,8 @@ import com.metamx.druid.StorageAdapter;
import
com.metamx.druid.index.v1.QueryableIndexStorageAdapter
;
import
org.joda.time.Interval
;
import
java.io.IOException
;
/**
*/
public
class
QueryableIndexSegment
implements
Segment
...
...
@@ -59,4 +61,11 @@ public class QueryableIndexSegment implements Segment
{
return
new
QueryableIndexStorageAdapter
(
index
);
}
@Override
public
void
close
()
throws
IOException
{
// this is kinda nasty
index
.
close
();
}
}
server/src/main/java/com/metamx/druid/index/ReferenceCountingSegment.java
0 → 100644
浏览文件 @
584ccac8
/*
* Druid - a distributed column store.
* Copyright (C) 2012 Metamarkets Group Inc.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*/
package
com.metamx.druid.index
;
import
com.metamx.druid.StorageAdapter
;
import
com.metamx.emitter.EmittingLogger
;
import
org.joda.time.Interval
;
import
java.io.Closeable
;
import
java.io.IOException
;
import
java.util.concurrent.atomic.AtomicBoolean
;
public
class
ReferenceCountingSegment
implements
Segment
{
private
static
final
EmittingLogger
log
=
new
EmittingLogger
(
ReferenceCountingSegment
.
class
);
private
final
Segment
baseSegment
;
private
final
Object
lock
=
new
Object
();
private
volatile
int
numReferences
=
0
;
private
volatile
boolean
isClosed
=
false
;
public
ReferenceCountingSegment
(
Segment
baseSegment
)
{
this
.
baseSegment
=
baseSegment
;
}
public
Segment
getBaseSegment
()
{
synchronized
(
lock
)
{
if
(!
isClosed
)
{
return
baseSegment
;
}
return
null
;
}
}
public
int
getNumReferences
()
{
return
numReferences
;
}
public
boolean
isClosed
()
{
return
isClosed
;
}
@Override
public
String
getIdentifier
()
{
synchronized
(
lock
)
{
if
(!
isClosed
)
{
return
baseSegment
.
getIdentifier
();
}
return
null
;
}
}
@Override
public
Interval
getDataInterval
()
{
synchronized
(
lock
)
{
if
(!
isClosed
)
{
return
baseSegment
.
getDataInterval
();
}
return
null
;
}
}
@Override
public
QueryableIndex
asQueryableIndex
()
{
synchronized
(
lock
)
{
if
(!
isClosed
)
{
return
baseSegment
.
asQueryableIndex
();
}
return
null
;
}
}
@Override
public
StorageAdapter
asStorageAdapter
()
{
synchronized
(
lock
)
{
if
(!
isClosed
)
{
return
baseSegment
.
asStorageAdapter
();
}
return
null
;
}
}
@Override
public
void
close
()
throws
IOException
{
synchronized
(
lock
)
{
if
(!
isClosed
)
{
if
(
numReferences
>
0
)
{
decrement
();
}
else
{
baseSegment
.
close
();
isClosed
=
true
;
}
}
}
}
public
Closeable
increment
()
{
synchronized
(
lock
)
{
if
(!
isClosed
)
{
numReferences
++;
final
AtomicBoolean
decrementOnce
=
new
AtomicBoolean
(
false
);
return
new
Closeable
()
{
@Override
public
void
close
()
throws
IOException
{
if
(
decrementOnce
.
compareAndSet
(
false
,
true
))
{
decrement
();
}
}
};
}
return
null
;
}
}
private
void
decrement
()
{
synchronized
(
lock
)
{
if
(!
isClosed
)
{
if
(--
numReferences
<
0
)
{
try
{
close
();
}
catch
(
Exception
e
)
{
log
.
error
(
"Unable to close queryable index %s"
,
getIdentifier
());
}
}
}
}
}
}
\ No newline at end of file
server/src/main/java/com/metamx/druid/index/ReferenceCountingSequence.java
0 → 100644
浏览文件 @
584ccac8
/*
* Druid - a distributed column store.
* Copyright (C) 2012 Metamarkets Group Inc.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*/
package
com.metamx.druid.index
;
import
com.metamx.common.guava.ResourceClosingYielder
;
import
com.metamx.common.guava.Sequence
;
import
com.metamx.common.guava.Yielder
;
import
com.metamx.common.guava.YieldingAccumulator
;
import
com.metamx.common.guava.YieldingSequenceBase
;
/**
*/
public
class
ReferenceCountingSequence
<
T
>
extends
YieldingSequenceBase
<
T
>
{
private
final
Sequence
<
T
>
baseSequence
;
private
final
ReferenceCountingSegment
segment
;
public
ReferenceCountingSequence
(
Sequence
<
T
>
baseSequence
,
ReferenceCountingSegment
segment
)
{
this
.
baseSequence
=
baseSequence
;
this
.
segment
=
segment
;
}
@Override
public
<
OutType
>
Yielder
<
OutType
>
toYielder
(
OutType
initValue
,
YieldingAccumulator
<
OutType
,
T
>
accumulator
)
{
return
new
ResourceClosingYielder
<
OutType
>(
baseSequence
.
toYielder
(
initValue
,
accumulator
),
segment
.
increment
());
}
}
\ No newline at end of file
server/src/main/java/com/metamx/druid/index/Segment.java
浏览文件 @
584ccac8
...
...
@@ -22,9 +22,11 @@ package com.metamx.druid.index;
import
com.metamx.druid.StorageAdapter
;
import
org.joda.time.Interval
;
import
java.io.Closeable
;
/**
*/
public
interface
Segment
public
interface
Segment
extends
Closeable
{
public
String
getIdentifier
();
public
Interval
getDataInterval
();
...
...
server/src/main/java/com/metamx/druid/loading/QueryableIndexFactory.java
浏览文件 @
584ccac8
...
...
@@ -22,6 +22,7 @@ package com.metamx.druid.loading;
import
com.metamx.druid.index.QueryableIndex
;
import
java.io.File
;
import
java.io.IOException
;
/**
*/
...
...
server/src/test/java/com/metamx/druid/coordination/ServerManagerTest.java
浏览文件 @
584ccac8
此差异已折叠。
点击以展开。
server/src/test/java/com/metamx/druid/index/ReferenceCountingSegmentTest.java
0 → 100644
浏览文件 @
584ccac8
/*
* Druid - a distributed column store.
* Copyright (C) 2012 Metamarkets Group Inc.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*/
package
com.metamx.druid.index
;
import
com.google.common.base.Throwables
;
import
com.metamx.druid.StorageAdapter
;
import
junit.framework.Assert
;
import
org.joda.time.Interval
;
import
org.junit.Before
;
import
org.junit.Test
;
import
java.io.Closeable
;
import
java.io.IOException
;
import
java.util.concurrent.ExecutorService
;
import
java.util.concurrent.Executors
;
/**
*/
public
class
ReferenceCountingSegmentTest
{
private
ReferenceCountingSegment
segment
;
private
ExecutorService
exec
;
@Before
public
void
setUp
()
throws
Exception
{
segment
=
new
ReferenceCountingSegment
(
new
Segment
()
{
@Override
public
String
getIdentifier
()
{
throw
new
UnsupportedOperationException
();
}
@Override
public
Interval
getDataInterval
()
{
throw
new
UnsupportedOperationException
();
}
@Override
public
QueryableIndex
asQueryableIndex
()
{
throw
new
UnsupportedOperationException
();
}
@Override
public
StorageAdapter
asStorageAdapter
()
{
throw
new
UnsupportedOperationException
();
}
@Override
public
void
close
()
throws
IOException
{
}
}
);
exec
=
Executors
.
newSingleThreadExecutor
();
}
@Test
public
void
testMultipleClose
()
throws
Exception
{
Assert
.
assertFalse
(
segment
.
isClosed
());
final
Closeable
closeable
=
segment
.
increment
();
Assert
.
assertTrue
(
segment
.
getNumReferences
()
==
1
);
closeable
.
close
();
closeable
.
close
();
exec
.
submit
(
new
Runnable
()
{
@Override
public
void
run
()
{
try
{
closeable
.
close
();
}
catch
(
Exception
e
)
{
throw
Throwables
.
propagate
(
e
);
}
}
}
);
Assert
.
assertTrue
(
segment
.
getNumReferences
()
==
0
);
Assert
.
assertFalse
(
segment
.
isClosed
());
segment
.
close
();
segment
.
close
();
exec
.
submit
(
new
Runnable
()
{
@Override
public
void
run
()
{
try
{
segment
.
close
();
}
catch
(
Exception
e
)
{
throw
Throwables
.
propagate
(
e
);
}
}
}
);
Assert
.
assertTrue
(
segment
.
getNumReferences
()
==
0
);
Assert
.
assertTrue
(
segment
.
isClosed
());
segment
.
increment
();
segment
.
increment
();
segment
.
increment
();
Assert
.
assertTrue
(
segment
.
getNumReferences
()
==
0
);
segment
.
close
();
Assert
.
assertTrue
(
segment
.
getNumReferences
()
==
0
);
}
}
server/src/test/java/com/metamx/druid/loading/CacheTestSegmentLoader.java
浏览文件 @
584ccac8
...
...
@@ -27,6 +27,7 @@ import com.metamx.druid.index.Segment;
import
org.joda.time.Interval
;
import
java.io.File
;
import
java.io.IOException
;
import
java.util.Map
;
/**
...
...
@@ -68,6 +69,11 @@ public class CacheTestSegmentLoader implements SegmentLoader
{
throw
new
UnsupportedOperationException
();
}
@Override
public
void
close
()
throws
IOException
{
}
};
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录