Metrics partition_input_bytes for data generator source is not recorded.

[MrCroxx]

Describe the bug

When using nexmark source generator, Source Throughput(MB/s) is not shown on Grafana, which queries metrics partition_input_bytes.

partition_input_bytes is updated by into_chunk_stream in src/connector/src/source/common.rs, but nexmark source generator doesn't call the function.

[MrCroxx]

The row-level source throughput metrics source_output_row_count(stream_source_output_rows_counts) is held by StreamingMetrics but byte-level source throughput metrics partition_input_count is held by SourceMetrics. Is this by design? @tabVersion

[tabVersion]

The row-level source throughput metrics source_output_row_count(stream_source_output_rows_counts) is held by StreamingMetrics but byte-level source throughput metrics partition_input_count is held by SourceMetrics. Is this by design? @tabVersion

Sorry for late reply. I think no, the metrics are added separately. I will submit a fix.

partition_input_bytes is updated by into_chunk_stream in src/connector/src/source/common.rs, but nexmark source generator doesn't call the function

IIRC, necmark source generates stream chunk directly, shall we use the chunk's size for this metric?

[MrCroxx]

IIRC, necmark source generates stream chunk directly, shall we use the chunk's size for this metric?

It's okay for me. But I think it will be better to use the same function for source data gen and normal source.