code: private void handleLogSealed(Segment segment) {
          sealedSegmentQueue.add(segment);
          retransmitPool.execute(() -> {
              Retry.indefinitelyWithExpBackoff(config.getInitalBackoffMillis(), config.getBackoffMultiple(),
                                               config.getMaxBackoffMillis(),
                                               t -> log.error("Encountered excemption when handeling a sealed segment: ", t))
                   .run(() -> {
                       /*
                        * Using writeSealLock prevents concurrent segmentSealedCallback for different segments
                        * from being invoked concurrently, or concurrently with write.
                        * 
                        * By calling flush while the write lock is held we can ensure that any inflight
                        * entries that will succeed in being written to a new segment are written and any
                        * segmentSealedCallbacks that will be called happen before the next write is invoked.
                        */
                       synchronized (writeSealLock) {
                           Segment toSeal = sealedSegmentQueue.poll();
                           log.info("Sealing segment {} ", toSeal);
                           while (toSeal != null) {
                               resend(selector.refreshSegmentEventWritersUponSealed(toSeal, segmentSealedCallBack));
                               /* In the case of segments merging Flush ensures there can't be anything left
                                * inflight that will need to be resent to the new segment when the write lock
                                * is released. (To preserve order)
                                */
                               for (SegmentOutputStream writer : selector.getWriters()) {
                                   try {
                                       writer.write(PendingEvent.withoutHeader(null, ByteBufferUtils.EMPTY, null));
                                       writer.flush();
                                   } catch (SegmentSealedException e) {
                                       // Segment sealed exception observed during a flush. Re-run flush on all the
                                       // available writers.
                                       log.info("Flush on segment {} failed due to {}, it will be retried.", writer.getSegmentName(), e.getMessage());
                                   }
                               }
                               toSeal = sealedSegmentQueue.poll();
                               log.info("Sealing another segment {} ", toSeal);
                           }
                       }
                       return null;
                   });
          });
      }
docs: If a log sealed is encountered, we need to 1. Find the new segments to write to. 2. For each outstanding
  message find which new segment it should go to and send it there.