@Override
    public Row deserialize(byte[] message) throws IOException {
        try {
            if (numInRecord.getCount() % dataPrintFrequency == 0) {
                LOG.info("receive source data:" + new String(message, StandardCharsets.UTF_8));
            }
            numInRecord.inc();
            numInBytes.inc(message.length);
            beforeDeserialize();
            Row row = deserializationSchema.deserialize(message);
            afterDeserialize();
            numInResolveRecord.inc();
            return row;
        } catch (Exception e) {
            //add metric of dirty data
            dirtyDataManager.execute();
            dirtyDataManager.collectDirtyData(
                    new String(message), ExceptionTrace.traceOriginalCause(e));
            dirtyDataCounter.inc();
            return null;
        }
    }
rps Record Per Second: deserialize data and out record num
core/src/main/java/com/dtstack/flink/sql/format/DeserializationMetricWrapper.java