UnboundedFeedbackLogger implements Closeable { public void commit() { try { flushToKeyedStateOutputStream(); } catch (IOException e) { throw new RuntimeException(e); } finally { keyGroupStreams.clear(); IOUtils.closeQuietly(snapshotLease); snapshotLease = null; keyedStateOutputStream = null; } } @Inject  UnboundedFeedbackLogger(
      @Label("key-group-supplier") Supplier<KeyGroupStream<T>> supplier,
      @Label("key-group-assigner") ToIntFunction<T> keyGroupAssigner,
      @Label("checkpoint-stream-ops") CheckpointedStreamOperations ops,
      @Label("envelope-serializer") TypeSerializer<T> serializer); void startLogging(OutputStream keyedStateCheckpointOutputStream); void append(T message); void commit(); void replyLoggedEnvelops(InputStream rawKeyedStateInputs, FeedbackConsumer<T> consumer); @Override void close();  }
@Test(expected = IllegalStateException.class) public void commitWithoutStartLoggingShouldBeIllegal() { UnboundedFeedbackLogger logger = instanceUnderTest(128, 1); logger.commit(); }