@Test(expected = AzureException.class)
  public void  testRespectFlushTimeout() throws Exception, InterruptedException {
    // get the threadpool to be the exactly the same as that passed down to AzureBlobOutputStream from AzureBlobSystemProducer
    threadPool = new ThreadPoolExecutor(1, 1, 60,
        TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>(1), new ThreadPoolExecutor.CallerRunsPolicy());

    // set a very small flushtimeout of 10ms to avoid test taking too long to complete
    azureBlobOutputStream = spy(new AzureBlobOutputStream(mockBlobAsyncClient, threadPool, mockMetrics,
        blobMetadataGeneratorFactory, blobMetadataGeneratorConfig, FAKE_STREAM,
        10, THRESHOLD, mockByteArrayOutputStream, mockCompression));

    doNothing().when(azureBlobOutputStream).clearAndMarkClosed();
    doReturn(mockBlobMetadataGenerator).when(azureBlobOutputStream).getBlobMetadataGenerator();
    when(mockCompression.compress(BYTES)).thenReturn(COMPRESSED_BYTES, COMPRESSED_BYTES, COMPRESSED_BYTES, COMPRESSED_BYTES);

    // create a latch to mimic uploads getting stuck
    // and hence unable to honor flush timeout without the fix in stageblock
    // fix in stageBlock = subscribeOn(Schedulers.boundedElastic()).block(flushtimeout)
    CountDownLatch latch = new CountDownLatch(1);
    doAnswer(invocation -> {

      String blockid = invocation.getArgumentAt(0, String.class);
      return Mono.just(1).map(integer -> {
        try {
          LOG.info("For block id = " + blockid + " start waiting on the countdown latch ");
          // start indefinite stuck -> mimic upload stuck
          latch.await();
          // below log will never be reached
          LOG.info("For block id = " + blockid + " done waiting on the countdown latch ");
        } catch (Exception e) {
          LOG.info("For block id = " + blockid + " an exception was caught " + e);
        }
        return "One";
      });
    }).when(azureBlobOutputStream).invokeBlobClientStageBlock(anyString(), anyObject(), anyInt());

    doAnswer(invocation -> {
      LOG.info("commit block ");
      return null;
    }).when(azureBlobOutputStream).commitBlob(anyObject(), anyMap());

    azureBlobOutputStream.write(BYTES, 0, THRESHOLD / 2);
    azureBlobOutputStream.write(BYTES, THRESHOLD / 2, THRESHOLD / 2);
    azureBlobOutputStream.write(BYTES, 0, THRESHOLD / 2);
    azureBlobOutputStream.write(BYTES, THRESHOLD / 2, THRESHOLD / 2);
    azureBlobOutputStream.write(BYTES, 0, THRESHOLD / 2);
    azureBlobOutputStream.write(BYTES, THRESHOLD / 2, THRESHOLD / 2);
    // close will wait for all pending uploads to finish
    // since the uploads are "stuck" (waiting for latch countdown), flushtimeout will get triggered
    // and throw an exception saying upload failed.
    azureBlobOutputStream.close();
  }
Test to ensure that flush timeout is respected even if the block upload to azure is stuck/ taking longer than flush timeout
a countdown latch is used to mimic the upload to azure stuck
if flush timeout is respected then an exception is thrown when the flushtimeout_ms duration expires
else if timeout is not respected (aka bug is not fixed) then no exception is thrown and test hangs
In this test, the flush timeout is chosen to be 10 milliseconds, at the end of which, an AzureException of upload failed is thrown.
@throws Exception
@throws InterruptedException
samza-azure/src/test/java/org/apache/samza/system/azureblob/avro/TestAzureBlobOutputStream.java