@Override
  public synchronized void publish(LogRecord record) {
    DataflowExecutionState currrentDataflowState = null;
    ExecutionState currrentState = ExecutionStateTracker.getCurrentExecutionState();
    if (currrentState instanceof DataflowExecutionState) {
      currrentDataflowState = (DataflowExecutionState) currrentState;
    }
    // It's okay to pass in the null state, publish() handles and tests this.
    publish(currrentDataflowState, record);
  }
@Test
  public void testOutputStreamRollover() throws IOException {
    DataflowWorkerLoggingMDC.setJobId("testJobId");

    ByteArrayOutputStream first = new ByteArrayOutputStream();
    ByteArrayOutputStream second = new ByteArrayOutputStream();

    LogRecord record = createLogRecord("test.message", null /* throwable */);
    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 /* sizelimit */);

    // Using |expected|+1 for size limit means that we will rollover after writing 2 log messages.
    // We thus expect to see 2 messsages written to 'first' and 1 message to 'second',

    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));
  }