Flink Process Element

Process.java

DataStream stream = ...; // Kafka stream consumer08 (version)
stream.keyBy("personId");    
stream.process(new ProcessAlerts());    

/**
 * The data type stored in the state
 */
public class PersonDetail {
    public int personId;
    public long count;
    public long lastModified;
}


@Slf4j
public class ProcessAlerts extends ProcessFunction<PersonStream, Object> {
  private ValueState<PersonDetail> state;
  
    @Override
    public void open(Configuration config) {
      ValueStateDescriptor<PersonDetail> descriptor = new ValueStateDescriptor<>("personState", TypeInformation.of(new TypeHint<PersonDetail>() {}));       
      state = getRuntimeContext().getState(descriptor);
    }  
  
  public void processElement(PersonStream ele, Context ctx, Collector<Object> out) {
    try {
      PersonDetail currentState = state.value();
      currentState.count++;
      currentState.lastModified = System.currentTimeMillis();
      state.update(currentState);
    } catch (e) {
      log.error("Error occurred while processing payload {}, Exception : {}", ele, e);
      log.error("{}", e);  
    }
  }
}