[opt-36803][file] Adjust fileSource metric logic.
@@ -196,13 +196,15 @@ public class FileSource extends AbstractRichFunction implements IStreamSourceGen
@Override
public void run(SourceContext<Row> ctx) throws Exception {
- AtomicInteger currentLine = new AtomicInteger(0);
- String line;
initMetric();
+ AtomicInteger currentLine = new AtomicInteger(0);
+ String line = "";
inputStream = getInputStream(fileUri);
bufferedReader = new BufferedReader(new InputStreamReader(inputStream, charset));
+ try {
while (running.get()) {
+ try {
line = bufferedReader.readLine();
if (line == null) {
running.compareAndSet(true, false);
@@ -214,7 +216,6 @@ public class FileSource extends AbstractRichFunction implements IStreamSourceGen
continue;
}
- try {
numInRecord.inc();
Row row = deserializationSchema.deserialize(line.getBytes());
if (row == null) {
@@ -222,6 +223,7 @@ public class FileSource extends AbstractRichFunction implements IStreamSourceGen
}
ctx.collect(row);
numInResolveRecord.inc();
+ }
} catch (IOException e) {
if (errorCounter.getCount() % 1000 == 0) {
LOG.error("Deserialize error! Record: " + line);
@@ -230,9 +232,10 @@ public class FileSource extends AbstractRichFunction implements IStreamSourceGen
errorCounter.inc();
}
}
- }
+ } finally {
ThreadUtil.sleepSeconds(METRIC_WAIT_TIME);
}
+ }
@Override
public void cancel() {
