XComp commented on code in PR #21467:
URL: https://github.com/apache/flink/pull/21467#discussion_r1055246893
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java:
##########
@@ -202,8 +203,8 @@ static void setupAll() {
rpcService = new TestingRpcService();
fastHeartbeatServices =
- new HeartbeatServices(fastHeartbeatInterval,
fastHeartbeatTimeout, -1);
- heartbeatServices = new HeartbeatServices(heartbeatInterval,
heartbeatTimeout, 1);
+ new HeartbeatServicesImpl(fastHeartbeatInterval,
fastHeartbeatTimeout, -1);
Review Comment:
Could we use `HeartbeatMonitorImpl.FAILED_RPC_DETECTION_DISABLED` here as
instread?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/HeartbeatMonitorImpl.java:
##########
@@ -39,6 +39,8 @@
private static final Logger LOG =
LoggerFactory.getLogger(HeartbeatMonitorImpl.class);
+ public static final int FAILED_RPC_DETECTION_DISABLED = -1;
Review Comment:
Initially, I wanted to suggest that we could reduce the visibility of this
field as well after moving `RecordingHeartbeatServices`. But then I noticed
that we actually should have this variable covered in the configuration docs
due to it being set in `HeartbeatManagerServices.fromConfiguration`. It looks
like we could use the field as well in
[HeartbeatManagerOptions:71](https://github.com/apache/flink/blob/e921c4c34b5497f4ba723ddae58750f6778069fa/flink-core/src/main/java/org/apache/flink/configuration/HeartbeatManagerOptions.java#L71)
##########
flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java:
##########
@@ -2896,7 +2899,9 @@ private TaskExecutorTestingContext
createTaskExecutorTestingContext(
.setTaskStateManager(stateStoresManager)
.setTaskChangelogStoragesManager(changelogStoragesManager)
.build(),
- HEARTBEAT_SERVICES,
+ // prevent heartbeat timeouts from failing the tests
focused on other
+ // aspects
+ HeartbeatServices.noOp(),
Review Comment:
Did we check that none of the other tests are affected by this change?
Besides `TaskExecutorTest.testSharedResourcesLifecycle`, also
`testDynamicSlotAllocation`, `testTaskSlotTableTerminationOnShutdown` and
`testReleasingJobResources` are using this utility method and this
configuration. I'm wondering whether we should make this a method parameter and
just use it for `testSharedResourcesLifecycle` to be on the safe side?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/NoOpHeartbeatServices.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.heartbeat;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.util.concurrent.ScheduledExecutor;
+
+import org.slf4j.Logger;
+
+/** {@link HeartbeatServices} implementation which does nothing. */
+public class NoOpHeartbeatServices implements HeartbeatServices {
Review Comment:
I'm not sure about the `NoOpHeartbeatServices` replacing
`HeartbeatServicesImpl` that have `failedRpcRequestsUntilUnreachable=-1` set:
The heartbeat services might still share payload even if no error is reported
due to missing messages. :thinking:
##########
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/NoOpHeartbeatServices.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.heartbeat;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.util.concurrent.ScheduledExecutor;
+
+import org.slf4j.Logger;
+
+/** {@link HeartbeatServices} implementation which does nothing. */
+public class NoOpHeartbeatServices implements HeartbeatServices {
Review Comment:
It still feels like we should move it to the test code. But I see your
point. So, no strong feeling here to push for this change. I'll leave it up to
you.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat/HeartbeatServices.java:
##########
@@ -100,37 +62,31 @@ public <I, O> HeartbeatManager<I, O>
createHeartbeatManager(
* @param <O> Type of the outgoing payload
* @return A new HeartbeatManager instance which actively sends heartbeats
*/
- public <I, O> HeartbeatManager<I, O> createHeartbeatManagerSender(
+ <I, O> HeartbeatManager<I, O> createHeartbeatManagerSender(
ResourceID resourceId,
HeartbeatListener<I, O> heartbeatListener,
ScheduledExecutor mainThreadExecutor,
- Logger log) {
-
- return new HeartbeatManagerSenderImpl<>(
- heartbeatInterval,
- heartbeatTimeout,
- failedRpcRequestsUntilUnreachable,
- resourceId,
- heartbeatListener,
- mainThreadExecutor,
- log);
- }
+ Logger log);
/**
* Creates an HeartbeatServices instance from a {@link Configuration}.
*
* @param configuration Configuration to be used for the HeartbeatServices
creation
* @return An HeartbeatServices instance created from the given
configuration
*/
- public static HeartbeatServices fromConfiguration(Configuration
configuration) {
+ static HeartbeatServices fromConfiguration(Configuration configuration) {
Review Comment:
ok, I see your point. :+1:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
