Hi All how do I use `CheckpointListener` to listen to comple Apache Flink #troubleshooting

Hi All, how do I use `CheckpointListener` to liste...

Guruguha Marur Sreenivasa

03/30/2023, 9:18 PM

Hi All, how do I use

CheckpointListener

to listen to completed checkpoints? I'm unable to register this with my stream environment. Can someone please help here? Below is my listener code:

Copy code

public class CheckpointCoordinator implements CheckpointListener {
  @Override
  public void notifyCheckpointComplete(long checkpointId) throws Exception {
    <http://log.info|log.info>("Checkpoint {} completed successfully.", checkpointId);
    // add custom logic here
  }

  @Override
  public void notifyCheckpointAborted(long checkpointId) throws Exception {
    CheckpointListener.super.notifyCheckpointAborted(checkpointId);
  }
}

How do I register this?

sap1ens

03/30/2023, 9:37 PM

AFAIK you can’t just add it to a random class, it has to be one of the Flink operators, e.g.

RichSinkFunction

Guruguha Marur Sreenivasa

03/30/2023, 9:37 PM

I tried this - I get this error when trying to submit the application:

ERROR!!! Coordinator Provider for node Source: ....  is not serializable.

Guruguha Marur Sreenivasa

03/30/2023, 9:40 PM

I added it to a

KeyedProcessFunction

Copy code

public class MultiModeProcessFilter extends KeyedProcessFunction<String, VLFRecord, FilteredRecord> implements CheckpointListener {
...

  @Override
  public void notifyCheckpointComplete(long checkpointId) throws Exception {
    <http://log.info|log.info>("**checkpoint completed ** {}", checkpointId);
    // custom logic handler
  }
...
}

sap1ens

03/30/2023, 9:47 PM

What’s the full exception that you see after that?

Guruguha Marur Sreenivasa

03/30/2023, 9:59 PM

Sorry about that - it was something internal. Nothing to do with adding the listener.

Open in Slack

Previous Next