Skip to content

[BEAM-9399] Ensure that empty messages are not flushed to handler. #11351

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lukecwik merged 1 commit into from
Apr 9, 2020
Merged

[BEAM-9399] Ensure that empty messages are not flushed to handler. #11351

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 22 additions & 22 deletions .../org/apache/beam/runners/dataflow/worker/logging/JulHandlerPrintStreamAdapterFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,13 +91,16 @@ public void write(int i) throws IOException {

@Override
public void flush() {
publish(flushToString());
publishIfNonEmpty(flushToString());
}

private synchronized String flushToString() {
if (buffer.length() > 0 && buffer.charAt(buffer.length() - 1) == '\n') {
buffer.setLength(buffer.length() - 1);
}
if (buffer.length() == 0) {
return null;
}
String result = buffer.toString();
buffer.setLength(0);
return result;
Expand Down Expand Up @@ -163,9 +166,7 @@ public void write(byte[] a, int offset, int length) {
}
}
}
if (msg != null) {
publish(msg);
}
publishIfNonEmpty(msg);
}

@Override
Expand Down Expand Up @@ -214,7 +215,7 @@ public void print(char[] a) {
}
msg = flushToString();
}
publish(msg);
publishIfNonEmpty(msg);
}

@Override
Expand All @@ -228,7 +229,7 @@ public void print(String s) {
}
msg = flushToString();
}
publish(msg);
publishIfNonEmpty(msg);
}

@Override
Expand All @@ -248,7 +249,7 @@ public void println(boolean b) {
buffer.append(b);
msg = flushToString();
}
publish(msg);
publishIfNonEmpty(msg);
}

@Override
Expand All @@ -258,7 +259,7 @@ public void println(char c) {
buffer.append(c);
msg = flushToString();
}
publish(msg);
publishIfNonEmpty(msg);
}

@Override
Expand All @@ -268,7 +269,7 @@ public void println(int i) {
buffer.append(i);
msg = flushToString();
}
publish(msg);
publishIfNonEmpty(msg);
}

@Override
Expand All @@ -278,7 +279,7 @@ public void println(long l) {
buffer.append(l);
msg = flushToString();
}
publish(msg);
publishIfNonEmpty(msg);
}

@Override
Expand All @@ -288,7 +289,7 @@ public void println(float f) {
buffer.append(f);
msg = flushToString();
}
publish(msg);
publishIfNonEmpty(msg);
}

@Override
Expand All @@ -298,7 +299,7 @@ public void println(double d) {
buffer.append(d);
msg = flushToString();
}
publish(msg);
publishIfNonEmpty(msg);
}

@Override
Expand All @@ -308,7 +309,7 @@ public void println(char[] a) {
buffer.append(a);
msg = flushToString();
}
publish(msg);
publishIfNonEmpty(msg);
}

@Override
Expand All @@ -318,7 +319,7 @@ public void println(String s) {
buffer.append(s);
msg = flushToString();
}
publish(msg);
publishIfNonEmpty(msg);
}

@Override
Expand All @@ -328,7 +329,7 @@ public void println(Object o) {
buffer.append(o);
msg = flushToString();
}
publish(msg);
publishIfNonEmpty(msg);
}

@Override
Expand All @@ -348,7 +349,7 @@ public PrintStream format(Locale locale, String format, Object... args) {
}
msg = flushToString();
}
publish(msg);
publishIfNonEmpty(msg);
return this;
}

Expand All @@ -370,15 +371,18 @@ public PrintStream append(CharSequence cs, int start, int limit) {
}
msg = flushToString();
}
publish(msg);
publishIfNonEmpty(msg);
return this;
}

// Note to avoid a deadlock, publish may never be called synchronized. See BEAM-9399.
private void publish(Level messageLevel, String message) {
private void publishIfNonEmpty(String message) {
checkState(
!Thread.holdsLock(this),
"BEAM-9399: publish should not be called with the lock as it may cause deadlock");
if (message == null || message.isEmpty()) {
return;
}
if (logger.isLoggable(messageLevel)) {
if (outputWarning.compareAndSet(false, true)) {
LogRecord log = new LogRecord(Level.WARNING, LOGGING_DISCLAIMER);
Expand All @@ -390,10 +394,6 @@ private void publish(Level messageLevel, String message) {
handler.publish(log);
}
}

private void publish(String message) {
publish(messageLevel, message);
}
}

/**
Expand Down
21 changes: 21 additions & 0 deletions .../apache/beam/runners/dataflow/worker/logging/JulHandlerPrintStreamAdapterFactoryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.beam.runners.dataflow.worker.logging;

import static org.apache.beam.runners.dataflow.worker.LogRecordMatcher.hasLogItem;
import static org.hamcrest.Matchers.blankOrNullString;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
Expand Down Expand Up @@ -132,6 +133,26 @@ public void testLogRawBytes() {
assertThat(handler.getLogs(), hasLogItem(msg + newlineMsg));
}

@Test
public void testNoEmptyMessages() {
try (PrintStream printStream = createPrintStreamAdapter()) {
printStream.println("blah");
printStream.print("\n");
printStream.flush();
printStream.println("");
printStream.flush();
printStream.print("");
printStream.flush();
byte[] bytes = "a".getBytes(Charset.defaultCharset());
printStream.write(bytes, 0, 0);
printStream.flush();
}

for (LogRecord log : handler.getLogs()) {
assertThat(log.getMessage(), not(blankOrNullString()));
}
}

private PrintStream createPrintStreamAdapter() {
return JulHandlerPrintStreamAdapterFactory.create(handler, LOGGER_NAME, Level.INFO);
}
Expand Down