### Question:
ZkBucketDataAccessor implements BucketDataAccessor, AutoCloseable { @Override public <T extends HelixProperty> boolean compressedBucketWrite(String rootPath, T value) throws IOException { DataUpdater<byte[]> lastWriteVersionUpdater = dataInZk -> { if (dataInZk == null || dataInZk.length == 0) { return "0".getBytes(); } String lastWriteVersionStr = new String(dataInZk); long lastWriteVersion = Long.parseLong(lastWriteVersionStr); lastWriteVersion++; return String.valueOf(lastWriteVersion).getBytes(); }; ZkBaseDataAccessor.AccessResult result = _zkBaseDataAccessor.doUpdate( rootPath + "/" + LAST_WRITE_KEY, lastWriteVersionUpdater, AccessOption.PERSISTENT); if (result._retCode != ZkBaseDataAccessor.RetCode.OK) { throw new HelixException( String.format("Failed to write the write version at path: %s!", rootPath)); } byte[] binaryVersion = (byte[]) result._updatedValue; String versionStr = new String(binaryVersion); final long version = Long.parseLong(versionStr); String versionedDataPath = rootPath + "/" + versionStr; byte[] serializedRecord = _zkSerializer.serialize(value.getRecord()); byte[] compressedRecord = GZipCompressionUtil.compress(serializedRecord); int numBuckets = (compressedRecord.length + _bucketSize - 1) / _bucketSize; List<String> paths = new ArrayList<>(); List<byte[]> buckets = new ArrayList<>(); int ptr = 0; int counter = 0; while (counter < numBuckets) { paths.add(versionedDataPath + "/" + counter); if (counter == numBuckets - 1) { buckets.add( Arrays.copyOfRange(compressedRecord, ptr, ptr + compressedRecord.length % _bucketSize)); } else { buckets.add(Arrays.copyOfRange(compressedRecord, ptr, ptr + _bucketSize)); } ptr += _bucketSize; counter++; } Map<String, String> metadata = ImmutableMap.of(BUCKET_SIZE_KEY, Integer.toString(_bucketSize), DATA_SIZE_KEY, Integer.toString(compressedRecord.length)); byte[] binaryMetadata = OBJECT_MAPPER.writeValueAsBytes(metadata); paths.add(versionedDataPath + "/" + METADATA_KEY); buckets.add(binaryMetadata); boolean[] success = _zkBaseDataAccessor.setChildren(paths, buckets, AccessOption.PERSISTENT); for (boolean s : success) { if (!s) { throw new HelixException( String.format("Failed to write the data buckets for path: %s", rootPath)); } } DataUpdater<byte[]> lastSuccessfulWriteVersionUpdater = dataInZk -> { if (dataInZk == null || dataInZk.length == 0) { return versionStr.getBytes(); } String lastWriteVersionStr = new String(dataInZk); long lastWriteVersion = Long.parseLong(lastWriteVersionStr); if (lastWriteVersion < version) { return versionStr.getBytes(); } else { return null; } }; if (!_zkBaseDataAccessor.update(rootPath + "/" + LAST_SUCCESSFUL_WRITE_KEY, lastSuccessfulWriteVersionUpdater, AccessOption.PERSISTENT)) { throw new HelixException(String .format("Failed to write the last successful write metadata at path: %s!", rootPath)); } updateGCTimer(rootPath, version); return true; } ZkBucketDataAccessor(String zkAddr, int bucketSize, long versionTTLms);  ZkBucketDataAccessor(String zkAddr); @Override boolean compressedBucketWrite(String rootPath, T value); @Override HelixProperty compressedBucketRead(String path,
      Class<T> helixPropertySubType); @Override void compressedBucketDelete(String path); @Override void disconnect(); @Override void close(); @Override void finalize();  }### Answer:
@Test public void testCompressedBucketWrite() throws IOException { Assert.assertTrue(_bucketDataAccessor.compressedBucketWrite(PATH, new HelixProperty(record))); }