Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
doujutun3207
flink
提交
73ab3175
F
flink
项目概览
doujutun3207
/
flink
与 Fork 源项目一致
从无法访问的项目Fork
通知
24
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
F
flink
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
73ab3175
编写于
6月 02, 2017
作者:
Z
zentol
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[FLINK-6825] Activate checkstyle for runtime/heartbeat
This closes #4067.
上级
1779a0eb
变更
8
隐藏空白更改
内联
并排
Showing
8 changed file
with
32 addition
and
28 deletion
+32
-28
flink-runtime/pom.xml
flink-runtime/pom.xml
+0
-1
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/HeartbeatListener.java
...org/apache/flink/runtime/heartbeat/HeartbeatListener.java
+2
-2
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/HeartbeatManager.java
.../org/apache/flink/runtime/heartbeat/HeartbeatManager.java
+2
-7
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/HeartbeatManagerImpl.java
.../apache/flink/runtime/heartbeat/HeartbeatManagerImpl.java
+13
-11
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/HeartbeatManagerSenderImpl.java
...e/flink/runtime/heartbeat/HeartbeatManagerSenderImpl.java
+1
-0
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/HeartbeatServices.java
...org/apache/flink/runtime/heartbeat/HeartbeatServices.java
+3
-2
flink-runtime/src/test/java/org/apache/flink/runtime/heartbeat/HeartbeatManagerTest.java
.../apache/flink/runtime/heartbeat/HeartbeatManagerTest.java
+7
-5
flink-runtime/src/test/java/org/apache/flink/runtime/heartbeat/TestingHeartbeatServices.java
...che/flink/runtime/heartbeat/TestingHeartbeatServices.java
+4
-0
未找到文件。
flink-runtime/pom.xml
浏览文件 @
73ab3175
...
...
@@ -434,7 +434,6 @@ under the License.
**/runtime/concurrent/**,
**/runtime/execution/**,
**/runtime/executiongraph/**,
**/runtime/heartbeat/**,
**/runtime/highavailability/**,
**/runtime/instance/**,
**/runtime/io/**,
...
...
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/HeartbeatListener.java
浏览文件 @
73ab3175
...
...
@@ -24,9 +24,9 @@ import org.apache.flink.runtime.concurrent.Future;
/**
* Interface for the interaction with the {@link HeartbeatManager}. The heartbeat listener is used
* for the following things:
*
<p>
*
* <ul>
* <
il>Notifications about heartbeat timeouts</il
>
* <
li>Notifications about heartbeat timeouts</li
>
* <li>Payload reports of incoming heartbeats</li>
* <li>Retrieval of payloads for outgoing heartbeats</li>
* </ul>
...
...
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/HeartbeatManager.java
浏览文件 @
73ab3175
...
...
@@ -21,13 +21,8 @@ package org.apache.flink.runtime.heartbeat;
import
org.apache.flink.runtime.clusterframework.types.ResourceID
;
/**
* A heartbeat manager has to be able to do the following things:
*
* <ul>
* <li>Monitor {@link HeartbeatTarget} and report heartbeat timeouts for this target</li>
* <li>Stop monitoring a {@link HeartbeatTarget}</li>
* </ul>
*
* A heartbeat manager has to be able to start/stop monitoring a {@link HeartbeatTarget}, and report heartbeat timeouts
* for this target.
*
* @param <I> Type of the incoming payload
* @param <O> Type of the outgoing payload
...
...
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/HeartbeatManagerImpl.java
浏览文件 @
73ab3175
...
...
@@ -24,9 +24,11 @@ import org.apache.flink.runtime.concurrent.ApplyFunction;
import
org.apache.flink.runtime.concurrent.Future
;
import
org.apache.flink.runtime.concurrent.ScheduledExecutor
;
import
org.apache.flink.util.Preconditions
;
import
org.slf4j.Logger
;
import
javax.annotation.concurrent.ThreadSafe
;
import
java.util.Collection
;
import
java.util.concurrent.ConcurrentHashMap
;
import
java.util.concurrent.Executor
;
...
...
@@ -47,27 +49,27 @@ import java.util.concurrent.atomic.AtomicReference;
@ThreadSafe
public
class
HeartbeatManagerImpl
<
I
,
O
>
implements
HeartbeatManager
<
I
,
O
>
{
/** Heartbeat timeout interval in milli seconds */
/** Heartbeat timeout interval in milli seconds
.
*/
private
final
long
heartbeatTimeoutIntervalMs
;
/** Resource ID which is used to mark one own's heartbeat signals */
/** Resource ID which is used to mark one own's heartbeat signals
.
*/
private
final
ResourceID
ownResourceID
;
/** Heartbeat listener with which the heartbeat manager has been associated */
/** Heartbeat listener with which the heartbeat manager has been associated
.
*/
private
final
HeartbeatListener
<
I
,
O
>
heartbeatListener
;
/** Executor service used to run heartbeat timeout notifications */
/** Executor service used to run heartbeat timeout notifications
.
*/
private
final
ScheduledExecutor
scheduledExecutor
;
protected
final
Logger
log
;
/** Map containing the heartbeat monitors associated with the respective resource ID */
/** Map containing the heartbeat monitors associated with the respective resource ID
.
*/
private
final
ConcurrentHashMap
<
ResourceID
,
HeartbeatManagerImpl
.
HeartbeatMonitor
<
O
>>
heartbeatTargets
;
/** Execution context used to run future callbacks */
/** Execution context used to run future callbacks
.
*/
private
final
Executor
executor
;
/** Running state of the heartbeat manager */
/** Running state of the heartbeat manager
.
*/
protected
volatile
boolean
stopped
;
public
HeartbeatManagerImpl
(
...
...
@@ -240,18 +242,18 @@ public class HeartbeatManagerImpl<I, O> implements HeartbeatManager<I, O> {
*/
static
class
HeartbeatMonitor
<
O
>
implements
Runnable
{
/** Resource ID of the monitored heartbeat target */
/** Resource ID of the monitored heartbeat target
.
*/
private
final
ResourceID
resourceID
;
/** Associated heartbeat target */
/** Associated heartbeat target
.
*/
private
final
HeartbeatTarget
<
O
>
heartbeatTarget
;
private
final
ScheduledExecutor
scheduledExecutor
;
/** Listener which is notified about heartbeat timeouts */
/** Listener which is notified about heartbeat timeouts
.
*/
private
final
HeartbeatListener
<?,
?>
heartbeatListener
;
/** Maximum heartbeat timeout interval */
/** Maximum heartbeat timeout interval
.
*/
private
final
long
heartbeatTimeoutIntervalMs
;
private
volatile
ScheduledFuture
<?>
futureTimeout
;
...
...
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/HeartbeatManagerSenderImpl.java
浏览文件 @
73ab3175
...
...
@@ -23,6 +23,7 @@ import org.apache.flink.runtime.concurrent.AcceptFunction;
import
org.apache.flink.runtime.concurrent.ApplyFunction
;
import
org.apache.flink.runtime.concurrent.Future
;
import
org.apache.flink.runtime.concurrent.ScheduledExecutor
;
import
org.slf4j.Logger
;
import
java.util.concurrent.Executor
;
...
...
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/HeartbeatServices.java
浏览文件 @
73ab3175
...
...
@@ -23,6 +23,7 @@ import org.apache.flink.configuration.HeartbeatManagerOptions;
import
org.apache.flink.runtime.clusterframework.types.ResourceID
;
import
org.apache.flink.runtime.concurrent.ScheduledExecutor
;
import
org.apache.flink.util.Preconditions
;
import
org.slf4j.Logger
;
/**
...
...
@@ -31,10 +32,10 @@ import org.slf4j.Logger;
*/
public
class
HeartbeatServices
{
/** Heartbeat interval for the created services */
/** Heartbeat interval for the created services
.
*/
protected
final
long
heartbeatInterval
;
/** Heartbeat timeout for the created services */
/** Heartbeat timeout for the created services
.
*/
protected
final
long
heartbeatTimeout
;
public
HeartbeatServices
(
long
heartbeatInterval
,
long
heartbeatTimeout
)
{
...
...
flink-runtime/src/test/java/org/apache/flink/runtime/heartbeat/HeartbeatManagerTest.java
浏览文件 @
73ab3175
...
...
@@ -26,6 +26,7 @@ import org.apache.flink.runtime.concurrent.ScheduledExecutorServiceAdapter;
import
org.apache.flink.runtime.concurrent.impl.FlinkCompletableFuture
;
import
org.apache.flink.runtime.util.DirectExecutorService
;
import
org.apache.flink.util.TestLogger
;
import
org.junit.Test
;
import
org.slf4j.Logger
;
import
org.slf4j.LoggerFactory
;
...
...
@@ -46,12 +47,14 @@ import static org.mockito.Matchers.anyLong;
import
static
org
.
mockito
.
Matchers
.
eq
;
import
static
org
.
mockito
.
Mockito
.
atLeast
;
import
static
org
.
mockito
.
Mockito
.
doReturn
;
import
static
org
.
mockito
.
Mockito
.
times
;
import
static
org
.
mockito
.
Mockito
.
mock
;
import
static
org
.
mockito
.
Mockito
.
times
;
import
static
org
.
mockito
.
Mockito
.
verify
;
import
static
org
.
mockito
.
Mockito
.
when
;
/**
* Tests for the {@link HeartbeatManager}.
*/
public
class
HeartbeatManagerTest
extends
TestLogger
{
private
static
final
Logger
LOG
=
LoggerFactory
.
getLogger
(
HeartbeatManagerTest
.
class
);
...
...
@@ -217,7 +220,7 @@ public class HeartbeatManagerTest extends TestLogger {
heartbeatListener2
,
new
DirectExecutorService
(),
new
ScheduledExecutorServiceAdapter
(
new
ScheduledThreadPoolExecutor
(
1
)),
LOG
);
;
LOG
);
heartbeatManager
.
monitorTarget
(
resourceID2
,
heartbeatManager2
);
heartbeatManager2
.
monitorTarget
(
resourceID
,
heartbeatManager
);
...
...
@@ -239,7 +242,7 @@ public class HeartbeatManagerTest extends TestLogger {
}
/**
* Tests that after unmonitoring a target, there won't be a timeout triggered
* Tests that after unmonitoring a target, there won't be a timeout triggered
.
*/
@Test
public
void
testTargetUnmonitoring
()
throws
InterruptedException
,
ExecutionException
{
...
...
@@ -265,7 +268,6 @@ public class HeartbeatManagerTest extends TestLogger {
Future
<
ResourceID
>
timeout
=
heartbeatListener
.
getTimeoutFuture
();
try
{
timeout
.
get
(
2
*
heartbeatTimeout
,
TimeUnit
.
MILLISECONDS
);
fail
(
"Timeout should time out."
);
...
...
flink-runtime/src/test/java/org/apache/flink/runtime/heartbeat/TestingHeartbeatServices.java
浏览文件 @
73ab3175
...
...
@@ -21,8 +21,12 @@ package org.apache.flink.runtime.heartbeat;
import
org.apache.flink.runtime.clusterframework.types.ResourceID
;
import
org.apache.flink.runtime.concurrent.ScheduledExecutor
;
import
org.apache.flink.util.Preconditions
;
import
org.slf4j.Logger
;
/**
* A {@link HeartbeatServices} that allows the injection of a {@link ScheduledExecutor}.
*/
public
class
TestingHeartbeatServices
extends
HeartbeatServices
{
private
final
ScheduledExecutor
scheduledExecutorToUse
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录