DataflowWorkerLoggingHandler extends Handler { @Override public synchronized void publish(LogRecord record) { DataflowExecutionState currrentDataflowState = null; ExecutionState currrentState = ExecutionStateTracker.getCurrentExecutionState(); if (currrentState instanceof DataflowExecutionState) { currrentDataflowState = (DataflowExecutionState) currrentState; } publish(currrentDataflowState, record); } DataflowWorkerLoggingHandler(String filename, long sizeLimit);   DataflowWorkerLoggingHandler(Supplier<OutputStream> factory, long sizeLimit); static String formatException(Throwable thrown); @Override synchronized void publish(LogRecord record); synchronized void publish(DataflowExecutionState currentExecutionState, LogRecord record); synchronized void publish(BeamFnApi.LogEntry logEntry); @Override boolean isLoggable(LogRecord record); @Override synchronized void flush(); @Override synchronized void close();  }
@Test public void testOutputStreamRollover() throws IOException { DataflowWorkerLoggingMDC.setJobId("testJobId"); ByteArrayOutputStream first = new ByteArrayOutputStream(); ByteArrayOutputStream second = new ByteArrayOutputStream(); LogRecord record = createLogRecord("test.message", null ); String expected = "{\"timestamp\":{\"seconds\":0,\"nanos\":1000000},\"severity\":\"INFO\"," + "\"message\":\"test.message\",\"thread\":\"2\",\"job\":\"testJobId\"," + "\"logger\":\"LoggerName\"}" + System.lineSeparator(); FixedOutputStreamFactory factory = new FixedOutputStreamFactory(first, second); DataflowWorkerLoggingHandler handler = new DataflowWorkerLoggingHandler(factory, expected.length() + 1 ); handler.publish(record); handler.publish(record); handler.publish(record); assertEquals(expected + expected, new String(first.toByteArray(), StandardCharsets.UTF_8)); assertEquals(expected, new String(second.toByteArray(), StandardCharsets.UTF_8)); }