carp84 commented on a change in pull request #13405:
URL: https://github.com/apache/flink/pull/13405#discussion_r492023475
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandler.java
##########
@@ -217,14 +218,14 @@ void snapshotState(
}
public void notifyCheckpointComplete(long checkpointId) throws
Exception {
- if (keyedStateBackend != null) {
-
keyedStateBackend.notifyCheckpointComplete(checkpointId);
+ if (keyedStateBackend != null && keyedStateBackend instanceof
CheckpointListener) {
Review comment:
```suggestion
if (keyedStateBackend instanceof CheckpointListener) {
```
Since the `instanceof` operator always returns `false` for `null`, there is
no need to have an additional null check.
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandler.java
##########
@@ -282,8 +283,8 @@ public void setCurrentKey(Object key) {
if (keyedStateBackend != null) {
try {
// need to work around type restrictions
- @SuppressWarnings("unchecked,rawtypes")
- AbstractKeyedStateBackend rawBackend =
(AbstractKeyedStateBackend) keyedStateBackend;
+ @SuppressWarnings("rawtypes")
+ CheckpointableKeyedStateBackend rawBackend =
keyedStateBackend;
rawBackend.setCurrentKey(key);
Review comment:
```suggestion
((CheckpointableKeyedStateBackend)
keyedStateBackend).setCurrentKey(key);
```
We could also remove "rawtypes" from the warning suppression of the method.
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandler.java
##########
@@ -217,14 +218,14 @@ void snapshotState(
}
public void notifyCheckpointComplete(long checkpointId) throws
Exception {
- if (keyedStateBackend != null) {
-
keyedStateBackend.notifyCheckpointComplete(checkpointId);
+ if (keyedStateBackend != null && keyedStateBackend instanceof
CheckpointListener) {
+ ((CheckpointListener)
keyedStateBackend).notifyCheckpointComplete(checkpointId);
}
}
public void notifyCheckpointAborted(long checkpointId) throws Exception
{
- if (keyedStateBackend != null) {
- keyedStateBackend.notifyCheckpointAborted(checkpointId);
+ if (keyedStateBackend != null && keyedStateBackend instanceof
CheckpointListener) {
Review comment:
```suggestion
if (keyedStateBackend instanceof CheckpointListener) {
```
Ditto.
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManager.java
##########
@@ -74,19 +74,16 @@
private final Map<String, InternalTimerServiceImpl<K, ?>> timerServices;
- private final boolean useLegacySynchronousSnapshots;
-
InternalTimeServiceManager(
- KeyGroupRange localKeyGroupRange,
- KeyContext keyContext,
- PriorityQueueSetFactory priorityQueueSetFactory,
- ProcessingTimeService processingTimeService, boolean
useLegacySynchronousSnapshots) {
Review comment:
It seems after the changes made for FLINK-16316 the
`useLegacySynchronousSnapshots` field here is never accessed? I'm also +1 for
keeping the private field and use it in `snapshotState` to save the cost of
`instanceof` when calling `requiresLegacySynchronousTimerSnapshots` for RocksDB
backend.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
