Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
Forever310
druid
提交
26569f4b
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,发现更多精彩内容 >>
提交
26569f4b
编写于
5月 10, 2014
作者:
N
nishantmonu51
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
fix jetty threads getting stuck
Fix Jetty threads getting stuck on exceptions
上级
48ae84ed
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
121 addition
and
45 deletion
+121
-45
pom.xml
pom.xml
+1
-1
server/src/main/java/io/druid/server/AsyncQueryForwardingServlet.java
...ain/java/io/druid/server/AsyncQueryForwardingServlet.java
+42
-34
server/src/test/java/io/druid/server/initialization/JettyTest.java
...c/test/java/io/druid/server/initialization/JettyTest.java
+78
-10
未找到文件。
pom.xml
浏览文件 @
26569f4b
...
...
@@ -79,7 +79,7 @@
<dependency>
<groupId>
com.metamx
</groupId>
<artifactId>
http-client
</artifactId>
<version>
0.9.
4
</version>
<version>
0.9.
5
</version>
</dependency>
<dependency>
<groupId>
com.metamx
</groupId>
...
...
server/src/main/java/io/druid/server/AsyncQueryForwardingServlet.java
浏览文件 @
26569f4b
...
...
@@ -42,6 +42,7 @@ import org.joda.time.DateTime;
import
javax.servlet.AsyncContext
;
import
javax.servlet.ServletException
;
import
javax.servlet.ServletOutputStream
;
import
javax.servlet.annotation.WebServlet
;
import
javax.servlet.http.HttpServlet
;
import
javax.servlet.http.HttpServletRequest
;
...
...
@@ -159,6 +160,15 @@ public class AsyncQueryForwardingServlet extends HttpServlet
return
ClientResponse
.
finished
(
obj
);
}
@Override
public
void
exceptionCaught
(
ClientResponse
<
OutputStream
>
clientResponse
,
Throwable
e
)
{
handleException
(
resp
,
asyncContext
,
e
);
}
};
asyncContext
.
start
(
...
...
@@ -176,23 +186,7 @@ public class AsyncQueryForwardingServlet extends HttpServlet
req
.
setAttribute
(
DISPATCHED
,
true
);
}
catch
(
Exception
e
)
{
if
(!
resp
.
isCommitted
())
{
resp
.
setStatus
(
500
);
resp
.
resetBuffer
();
if
(
out
==
null
)
{
out
=
resp
.
getOutputStream
();
}
if
(
ctx
!=
null
)
{
ctx
.
complete
();
}
out
.
write
((
e
.
getMessage
()
==
null
)
?
"Exception null"
.
getBytes
(
UTF8
)
:
e
.
getMessage
().
getBytes
(
UTF8
));
out
.
write
(
"\n"
.
getBytes
(
UTF8
));
}
resp
.
flushBuffer
();
handleException
(
resp
,
ctx
,
e
);
}
}
...
...
@@ -316,6 +310,15 @@ public class AsyncQueryForwardingServlet extends HttpServlet
return
ClientResponse
.
finished
(
obj
);
}
@Override
public
void
exceptionCaught
(
ClientResponse
<
OutputStream
>
clientResponse
,
Throwable
e
)
{
handleException
(
resp
,
asyncContext
,
e
);
}
};
asyncContext
.
start
(
...
...
@@ -333,23 +336,7 @@ public class AsyncQueryForwardingServlet extends HttpServlet
req
.
setAttribute
(
DISPATCHED
,
true
);
}
catch
(
Exception
e
)
{
if
(!
resp
.
isCommitted
())
{
resp
.
setStatus
(
500
);
resp
.
resetBuffer
();
if
(
out
==
null
)
{
out
=
resp
.
getOutputStream
();
}
out
.
write
((
e
.
getMessage
()
==
null
)
?
"Exception null"
.
getBytes
(
UTF8
)
:
e
.
getMessage
().
getBytes
(
UTF8
));
out
.
write
(
"\n"
.
getBytes
(
UTF8
));
}
resp
.
flushBuffer
();
if
(
ctx
!=
null
)
{
ctx
.
complete
();
}
handleException
(
resp
,
ctx
,
e
);
try
{
requestLogger
.
log
(
...
...
@@ -382,4 +369,25 @@ public class AsyncQueryForwardingServlet extends HttpServlet
}
return
String
.
format
(
"http://%s%s?%s"
,
host
,
requestURI
,
queryString
);
}
private
static
void
handleException
(
HttpServletResponse
resp
,
AsyncContext
ctx
,
Throwable
e
)
{
try
{
final
ServletOutputStream
out
=
resp
.
getOutputStream
();
if
(!
resp
.
isCommitted
())
{
resp
.
setStatus
(
500
);
resp
.
resetBuffer
();
out
.
write
((
e
.
getMessage
()
==
null
)
?
"Exception null"
.
getBytes
(
UTF8
)
:
e
.
getMessage
().
getBytes
(
UTF8
));
out
.
write
(
"\n"
.
getBytes
(
UTF8
));
}
if
(
ctx
!=
null
)
{
ctx
.
complete
();
}
resp
.
flushBuffer
();
}
catch
(
IOException
e1
)
{
Throwables
.
propagate
(
e1
);
}
}
}
server/src/test/java/io/druid/server/initialization/JettyTest.java
浏览文件 @
26569f4b
...
...
@@ -17,7 +17,7 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*/
package
io.druid
;
package
io.druid
.server.initialization
;
import
com.google.common.collect.Lists
;
import
com.google.common.util.concurrent.ListenableFuture
;
...
...
@@ -28,13 +28,14 @@ import com.google.inject.Module;
import
com.google.inject.servlet.GuiceFilter
;
import
com.metamx.common.lifecycle.Lifecycle
;
import
com.metamx.http.client.HttpClient
;
import
com.metamx.http.client.response.InputStreamResponseHandler
;
import
com.metamx.http.client.response.StatusResponseHandler
;
import
com.metamx.http.client.response.StatusResponseHolder
;
import
io.druid.guice.Jerseys
;
import
io.druid.guice.LazySingleton
;
import
io.druid.guice.annotations.Global
;
import
io.druid.initialization.Initialization
;
import
io.druid.server.initialization.JettyServerInitializer
;
import
org.apache.commons.io.IOUtils
;
import
org.eclipse.jetty.server.Handler
;
import
org.eclipse.jetty.server.Server
;
import
org.eclipse.jetty.server.handler.DefaultHandler
;
...
...
@@ -43,13 +44,22 @@ import org.eclipse.jetty.servlet.DefaultServlet;
import
org.eclipse.jetty.servlet.ServletContextHandler
;
import
org.eclipse.jetty.servlet.ServletHolder
;
import
org.eclipse.jetty.servlets.GzipFilter
;
import
org.junit.After
;
import
org.junit.Assert
;
import
org.junit.Before
;
import
org.junit.Ignore
;
import
org.junit.Test
;
import
javax.servlet.ServletOutputStream
;
import
javax.servlet.http.HttpServletResponse
;
import
javax.ws.rs.GET
;
import
javax.ws.rs.Path
;
import
javax.ws.rs.Produces
;
import
javax.ws.rs.core.Context
;
import
javax.ws.rs.core.Response
;
import
java.io.IOException
;
import
java.io.InputStream
;
import
java.io.StringWriter
;
import
java.net.URL
;
import
java.nio.charset.Charset
;
import
java.util.Random
;
...
...
@@ -61,6 +71,9 @@ import java.util.concurrent.atomic.AtomicLong;
public
class
JettyTest
{
private
Lifecycle
lifecycle
;
private
HttpClient
client
;
public
static
void
setProperties
()
{
System
.
setProperty
(
"druid.host"
,
"localhost:9999"
);
...
...
@@ -71,11 +84,9 @@ public class JettyTest
System
.
setProperty
(
"druid.global.http.readTimeout"
,
"PT1S"
);
}
@
Test
@Ignore
// this test will deadlock if it hits an issue, so ignored by default
public
void
testTimeouts
()
throws
Exception
@
Before
public
void
setup
()
throws
Exception
{
// test for request timeouts properly not locking up all threads
setProperties
();
Injector
injector
=
Initialization
.
makeInjectorWithModules
(
Initialization
.
makeStartupInjector
(),
Lists
.<
Object
>
newArrayList
(
...
...
@@ -86,17 +97,24 @@ public class JettyTest
{
binder
.
bind
(
JettyServerInitializer
.
class
).
to
(
JettyServerInit
.
class
).
in
(
LazySingleton
.
class
);
Jerseys
.
addResource
(
binder
,
SlowResource
.
class
);
Jerseys
.
addResource
(
binder
,
ExceptionResource
.
class
);
}
}
)
);
Lifecycle
lifecycle
=
injector
.
getInstance
(
Lifecycle
.
class
);
lifecycle
=
injector
.
getInstance
(
Lifecycle
.
class
);
// Jetty is Lazy Initialized do a getInstance
injector
.
getInstance
(
Server
.
class
);
lifecycle
.
start
();
ClientHolder
holder
=
injector
.
getInstance
(
ClientHolder
.
class
);
final
HttpClient
client
=
holder
.
getClient
();
client
=
holder
.
getClient
();
}
@Test
@Ignore
// this test will deadlock if it hits an issue, so ignored by default
public
void
testTimeouts
()
throws
Exception
{
// test for request timeouts properly not locking up all threads
final
Executor
executor
=
Executors
.
newFixedThreadPool
(
100
);
final
AtomicLong
count
=
new
AtomicLong
(
0
);
final
CountDownLatch
latch
=
new
CountDownLatch
(
1000
);
...
...
@@ -149,8 +167,35 @@ public class JettyTest
}
latch
.
await
();
lifecycle
.
stop
();
}
// Tests that threads are not stuck when partial chunk is not finalized
// https://bugs.eclipse.org/bugs/show_bug.cgi?id=424107
@Test
public
void
testChunkNotFinalized
()
throws
Exception
{
ListenableFuture
<
InputStream
>
go
=
client
.
get
(
new
URL
(
"http://localhost:9999/exception/exception"
)
)
.
go
(
new
InputStreamResponseHandler
());
try
{
StringWriter
writer
=
new
StringWriter
();
IOUtils
.
copy
(
go
.
get
(),
writer
,
"utf-8"
);
Assert
.
fail
(
"Should have thrown Exception"
);
}
catch
(
IOException
e
)
{
// Expected.
}
}
@After
public
void
teardown
()
{
lifecycle
.
stop
();
}
public
static
class
ClientHolder
...
...
@@ -206,4 +251,27 @@ public class JettyTest
return
Response
.
ok
(
"hello"
).
build
();
}
}
@Path
(
"/exception"
)
public
static
class
ExceptionResource
{
@GET
@Path
(
"/exception"
)
@Produces
(
"application/json"
)
public
Response
exception
(
@Context
HttpServletResponse
resp
)
throws
IOException
{
final
ServletOutputStream
outputStream
=
resp
.
getOutputStream
();
outputStream
.
println
(
"hello"
);
outputStream
.
flush
();
try
{
TimeUnit
.
MILLISECONDS
.
sleep
(
200
);
}
catch
(
InterruptedException
e
)
{
//
}
throw
new
IOException
();
}
}
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录