Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5737#discussion_r176207225
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/StringifiedAccumulatorResult.java
---
@@ -67,22 +72,29 @@ public String getValue() {
int i = 0;
for (Map.Entry<String, Accumulator<?, ?>> entry :
accs.entrySet()) {
- StringifiedAccumulatorResult result;
- Accumulator<?, ?> accumulator =
entry.getValue();
- if (accumulator != null) {
- Object localValue =
accumulator.getLocalValue();
- if (localValue != null) {
- result = new
StringifiedAccumulatorResult(entry.getKey(),
accumulator.getClass().getSimpleName(), localValue.toString());
- } else {
- result = new
StringifiedAccumulatorResult(entry.getKey(),
accumulator.getClass().getSimpleName(), "null");
- }
- } else {
- result = new
StringifiedAccumulatorResult(entry.getKey(), "null", "null");
- }
-
- results[i++] = result;
+ results[i++] =
stringifyAccumulatorResult(entry.getKey(), entry.getValue());
}
return results;
}
}
+
+ private static StringifiedAccumulatorResult
stringifyAccumulatorResult(String name, Accumulator<?, ?> accumulator) {
+ if (accumulator == null) {
+ return new StringifiedAccumulatorResult(name, "null",
"null");
+ } else {
+ Object localValue;
+ try {
+ localValue = accumulator.getLocalValue();
+ }
+ catch (RuntimeException exception) {
+ LOG.error("Failed to stringify accumulator",
exception);
--- End diff --
Maybe add `name` to log statement.
---
