@Override
  public byte[] compress(byte[] input) {
    Validate.notNull(input, "Input for compression is null");

    ByteArrayOutputStream byteArrayOutputStream = null;
    GZIPOutputStream gzipOutputStream = null;
    try {
      byteArrayOutputStream = new ByteArrayOutputStream(input.length);
      gzipOutputStream = new GZIPOutputStream(byteArrayOutputStream);
      gzipOutputStream.write(input);
      gzipOutputStream.close();
      return byteArrayOutputStream.toByteArray();

    } catch (IOException e) {
      throw new SamzaException("Failed to compress.", e);
    } finally {
      try {
        if (gzipOutputStream != null) {
          gzipOutputStream.close();
        }
        if (byteArrayOutputStream != null) {
          byteArrayOutputStream.close();
        }
      } catch (Exception e) {
        throw new SamzaException("Failed to close output streams during compression.", e);
      }
    }
  }
{@inheritDoc}
@throws SamzaException if compression fails
samza-azure/src/main/java/org/apache/samza/system/azureblob/compression/GzipCompression.java