BaseUploader { public void validate(List<Path> parts, PartitionerType partitionerType, int numShards) throws IOException { boolean hasNonEmptyPartition = false; HColumnDescriptor columnDescriptor = new HColumnDescriptor(); columnDescriptor.setBlockCacheEnabled(false); for (int shardIndex = 0; shardIndex < parts.size(); shardIndex++) { Path fileToBeValidated = parts.get(shardIndex); HFile.Reader reader = null; try { FileSystem fs = FileSystem.newInstance(fileToBeValidated.toUri(), conf); CacheConfig cc = new CacheConfig(conf, columnDescriptor); reader = HFile.createReader(fs, fileToBeValidated, cc); Partitioner partitioner = PartitionerFactory.getPartitioner(partitionerType); byte[] rowKey = reader.getFirstRowKey(); if (rowKey == null) { LOG.warn(String.format("empty partition %s", fileToBeValidated.toString())); reader.close(); continue; } hasNonEmptyPartition = true; BytesWritable key = new BytesWritable(rowKey); int partition = partitioner.getPartition(key, null, numShards); if (partition != shardIndex) { throw new IllegalArgumentException( String.format("wrong partition type %s for key %s in partition %d, expected %d", partitionerType.toString(), new String(key.getBytes()), shardIndex, partition) ); } } finally { if (reader != null) { reader.close(); } } } if (!hasNonEmptyPartition) { throw new IllegalArgumentException("all partitions are empty"); } } BaseUploader(TerrapinUploaderOptions uploaderOptions); void validate(List<Path> parts, PartitionerType partitionerType, int numShards); void upload(String clusterName, String fileSet, Options options);  }
@Test public void testValidateCascadingPartition() throws IOException { List<Path> parts = HFileGenerator.generateHFiles(fs, conf, tempFolder, PartitionerType.CASCADING, 10, 1000); uploader.validate(parts, PartitionerType.CASCADING, 10); }
@Test(expected = IllegalArgumentException.class) public void testValidateCascadingPartitionWithException() throws IOException { List<Path> parts = HFileGenerator.generateHFiles(fs, conf, tempFolder, PartitionerType.MODULUS, 10, 1000); uploader.validate(parts, PartitionerType.CASCADING, 10); }
@Test public void testValidateModulusPartition() throws IOException { List<Path> parts = HFileGenerator.generateHFiles(fs, conf, tempFolder, PartitionerType.MODULUS, 10, 1000); uploader.validate(parts, PartitionerType.MODULUS, 10); }
@Test(expected = IllegalArgumentException.class) public void testValidateModulusPartitionWithException() throws IOException { List<Path> parts = HFileGenerator.generateHFiles(fs, conf, tempFolder, PartitionerType.CASCADING, 10, 1000); uploader.validate(parts, PartitionerType.MODULUS, 10); }
@Test(expected = IllegalArgumentException.class) public void testAllEmptyPartitions() throws IOException { List<Path> parts = HFileGenerator.generateHFiles(fs, conf, tempFolder, PartitionerType.CASCADING, 10, 0); uploader.validate(parts, PartitionerType.MODULUS, 10); }