| /// * |
| // * Licensed to the Apache Software Foundation (ASF) under one or more |
| // * contributor license agreements. See the NOTICE file distributed with |
| // * this work for additional information regarding copyright ownership. |
| // * The ASF licenses this file to You under the Apache License, Version 2.0 |
| // * (the "License"); you may not use this file except in compliance with |
| // * the License. You may obtain a copy of the License at |
| // * |
| // * http://www.apache.org/licenses/LICENSE-2.0 |
| // * |
| // * Unless required by applicable law or agreed to in writing, software |
| // * distributed under the License is distributed on an "AS IS" BASIS, |
| // * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| // * See the License for the specific language governing permissions and |
| // * limitations under the License. |
| // */ |
| // |
| // package org.apache.accumulo.examples.filedata; |
| // |
| // import static org.junit.jupiter.api.Assertions.assertEquals; |
| // import static org.junit.jupiter.api.Assertions.assertFalse; |
| // import static org.junit.jupiter.api.Assertions.assertNotNull; |
| // import static org.junit.jupiter.api.Assertions.fail; |
| // |
| // import java.io.File; |
| // import java.io.IOException; |
| // import java.io.InputStream; |
| // import java.util.ArrayList; |
| // import java.util.List; |
| // import java.util.Map.Entry; |
| // |
| // import org.apache.accumulo.core.client.Accumulo; |
| // import org.apache.accumulo.core.client.AccumuloClient; |
| // import org.apache.accumulo.core.client.BatchWriter; |
| // import org.apache.accumulo.core.client.BatchWriterConfig; |
| // import org.apache.accumulo.core.conf.Property; |
| // import org.apache.accumulo.core.data.Key; |
| // import org.apache.accumulo.core.data.Mutation; |
| // import org.apache.accumulo.core.data.Value; |
| // import org.apache.accumulo.core.security.Authorizations; |
| // import org.apache.accumulo.core.security.ColumnVisibility; |
| // import org.apache.accumulo.harness.AccumuloClusterHarness; |
| // import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl; |
| // import org.apache.hadoop.conf.Configuration; |
| // import org.apache.hadoop.conf.Configured; |
| // import org.apache.hadoop.mapreduce.Job; |
| // import org.apache.hadoop.mapreduce.Mapper; |
| // import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat; |
| // import org.apache.hadoop.util.Tool; |
| // import org.apache.hadoop.util.ToolRunner; |
| // import org.junit.jupiter.api.AfterEach; |
| // import org.junit.jupiter.api.BeforeAll; |
| // import org.junit.jupiter.api.BeforeEach; |
| // import org.junit.jupiter.api.Test; |
| // |
| // import com.google.common.collect.ArrayListMultimap; |
| // import com.google.common.collect.Multimap; |
| // |
| // public class ChunkInputFormatIT extends AccumuloClusterHarness { |
| // @Override |
| // public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) { |
| // cfg.setProperty(Property.TSERV_NATIVEMAP_ENABLED, "false"); |
| // } |
| // |
| // // track errors in the map reduce job; jobs insert a dummy error for the map and cleanup tasks |
| /// (to |
| // // ensure test correctness), so error tests should check to see if there is at least one error |
| // // (could be more depending on the test) rather than zero |
| // private static final Multimap<String,AssertionError> assertionErrors = |
| /// ArrayListMultimap.create(); |
| // |
| // private static final Authorizations AUTHS = new Authorizations("A", "B", "C", "D"); |
| // |
| // private static List<Entry<Key,Value>> data; |
| // private static List<Entry<Key,Value>> baddata; |
| // |
| // private AccumuloClient client; |
| // private String tableName; |
| // |
| // @BeforeEach |
| // public void setupInstance() throws Exception { |
| // client = Accumulo.newClient().from(getClientProps()).build(); |
| // tableName = getUniqueNames(1)[0]; |
| // client.securityOperations().changeUserAuthorizations(client.whoami(), AUTHS); |
| // } |
| // |
| // @AfterEach |
| // public void teardown() { |
| // client.close(); |
| // } |
| // |
| // @BeforeAll |
| // public static void setupClass() { |
| // System.setProperty("hadoop.tmp.dir", System.getProperty("user.dir") + "/target/hadoop-tmp"); |
| // |
| // data = new ArrayList<>(); |
| // ChunkInputStreamIT.addData(data, "a", "refs", "ida\0ext", "A&B", "ext"); |
| // ChunkInputStreamIT.addData(data, "a", "refs", "ida\0name", "A&B", "name"); |
| // ChunkInputStreamIT.addData(data, "a", "~chunk", 100, 0, "A&B", "asdfjkl;"); |
| // ChunkInputStreamIT.addData(data, "a", "~chunk", 100, 1, "A&B", ""); |
| // ChunkInputStreamIT.addData(data, "b", "refs", "ida\0ext", "A&B", "ext"); |
| // ChunkInputStreamIT.addData(data, "b", "refs", "ida\0name", "A&B", "name"); |
| // ChunkInputStreamIT.addData(data, "b", "~chunk", 100, 0, "A&B", "qwertyuiop"); |
| // ChunkInputStreamIT.addData(data, "b", "~chunk", 100, 0, "B&C", "qwertyuiop"); |
| // ChunkInputStreamIT.addData(data, "b", "~chunk", 100, 1, "A&B", ""); |
| // ChunkInputStreamIT.addData(data, "b", "~chunk", 100, 1, "B&C", ""); |
| // ChunkInputStreamIT.addData(data, "b", "~chunk", 100, 1, "D", ""); |
| // baddata = new ArrayList<>(); |
| // ChunkInputStreamIT.addData(baddata, "c", "refs", "ida\0ext", "A&B", "ext"); |
| // ChunkInputStreamIT.addData(baddata, "c", "refs", "ida\0name", "A&B", "name"); |
| // } |
| // |
| // public static void entryEquals(Entry<Key,Value> e1, Entry<Key,Value> e2) { |
| // assertEquals(e1.getKey(), e2.getKey()); |
| // assertEquals(e1.getValue(), e2.getValue()); |
| // } |
| // |
| // public static class CIFTester extends Configured implements Tool { |
| // public static class TestMapper |
| // extends Mapper<List<Entry<Key,Value>>,InputStream,List<Entry<Key,Value>>,InputStream> { |
| // int count = 0; |
| // |
| // @Override |
| // protected void map(List<Entry<Key,Value>> key, InputStream value, Context context) |
| // throws IOException { |
| // String table = context.getConfiguration().get("MRTester_tableName"); |
| // assertNotNull(table); |
| // |
| // byte[] b = new byte[20]; |
| // int read; |
| // try (value) { |
| // switch (count) { |
| // case 0: |
| // assertEquals(key.size(), 2); |
| // entryEquals(key.get(0), data.get(0)); |
| // entryEquals(key.get(1), data.get(1)); |
| // assertEquals(read = value.read(b), 8); |
| // assertEquals(new String(b, 0, read), "asdfjkl;"); |
| // assertEquals(read = value.read(b), -1); |
| // break; |
| // case 1: |
| // assertEquals(key.size(), 2); |
| // entryEquals(key.get(0), data.get(4)); |
| // entryEquals(key.get(1), data.get(5)); |
| // assertEquals(read = value.read(b), 10); |
| // assertEquals(new String(b, 0, read), "qwertyuiop"); |
| // assertEquals(read = value.read(b), -1); |
| // break; |
| // default: |
| // fail(); |
| // } |
| // } catch (AssertionError e) { |
| // assertionErrors.put(table, e); |
| // } |
| // count++; |
| // } |
| // |
| // @Override |
| // protected void cleanup(Context context) { |
| // String table = context.getConfiguration().get("MRTester_tableName"); |
| // assertNotNull(table); |
| // |
| // try { |
| // assertEquals(2, count); |
| // } catch (AssertionError e) { |
| // assertionErrors.put(table, e); |
| // } |
| // } |
| // } |
| // |
| // public static class TestNoClose |
| // extends Mapper<List<Entry<Key,Value>>,InputStream,List<Entry<Key,Value>>,InputStream> { |
| // int count = 0; |
| // |
| // @Override |
| // protected void map(List<Entry<Key,Value>> key, InputStream value, Context context) |
| // throws IOException, InterruptedException { |
| // String table = context.getConfiguration().get("MRTester_tableName"); |
| // assertNotNull(table); |
| // |
| // byte[] b = new byte[5]; |
| // int read; |
| // try { |
| // switch (count) { |
| // case 0: |
| // assertEquals(read = value.read(b), 5); |
| // assertEquals(new String(b, 0, read), "asdfj"); |
| // break; |
| // default: |
| // fail(); |
| // } |
| // } catch (AssertionError e) { |
| // assertionErrors.put(table, e); |
| // } |
| // count++; |
| // try { |
| // context.nextKeyValue(); |
| // fail(); |
| // } catch (IOException ioe) { |
| // assertionErrors.put(table + "_map_ioexception", new AssertionError(toString(), ioe)); |
| // } |
| // } |
| // } |
| // |
| // public static class TestBadData |
| // extends Mapper<List<Entry<Key,Value>>,InputStream,List<Entry<Key,Value>>,InputStream> { |
| // @Override |
| // protected void map(List<Entry<Key,Value>> key, InputStream value, Context context) { |
| // String table = context.getConfiguration().get("MRTester_tableName"); |
| // assertNotNull(table); |
| // |
| // byte[] b = new byte[20]; |
| // try { |
| // assertEquals(key.size(), 2); |
| // entryEquals(key.get(0), baddata.get(0)); |
| // entryEquals(key.get(1), baddata.get(1)); |
| // } catch (AssertionError e) { |
| // assertionErrors.put(table, e); |
| // } |
| // try { |
| // assertFalse(value.read(b) > 0); |
| // try { |
| // fail(); |
| // } catch (AssertionError e) { |
| // assertionErrors.put(table, e); |
| // } |
| // } catch (Exception e) { |
| // // expected, ignore |
| // } |
| // try { |
| // value.close(); |
| // try { |
| // fail(); |
| // } catch (AssertionError e) { |
| // assertionErrors.put(table, e); |
| // } |
| // } catch (Exception e) { |
| // // expected, ignore |
| // } |
| // } |
| // } |
| // |
| // @SuppressWarnings("deprecation") |
| // @Override |
| // public int run(String[] args) throws Exception { |
| // if (args.length != 2) { |
| // throw new IllegalArgumentException( |
| // "Usage : " + CIFTester.class.getName() + " <table> <mapperClass>"); |
| // } |
| // |
| // String table = args[0]; |
| // assertionErrors.put(table, new AssertionError("Dummy")); |
| // assertionErrors.put(table + "_map_ioexception", new AssertionError("Dummy_ioexception")); |
| // getConf().set("MRTester_tableName", table); |
| // |
| // Job job = Job.getInstance(getConf()); |
| // job.setJobName(this.getClass().getSimpleName() + "_" + System.currentTimeMillis()); |
| // job.setJarByClass(this.getClass()); |
| // |
| // job.setInputFormatClass(ChunkInputFormat.class); |
| // |
| // ChunkInputFormat.setZooKeeperInstance(job, getCluster().getClientConfig()); |
| // ChunkInputFormat.setConnectorInfo(job, getAdminPrincipal(), getAdminToken()); |
| // ChunkInputFormat.setInputTableName(job, table); |
| // ChunkInputFormat.setScanAuthorizations(job, AUTHS); |
| // |
| // @SuppressWarnings("unchecked") |
| // Class<? extends Mapper<?,?,?,?>> forName = (Class<? extends Mapper<?,?,?,?>>) Class |
| // .forName(args[1]); |
| // job.setMapperClass(forName); |
| // job.setMapOutputKeyClass(Key.class); |
| // job.setMapOutputValueClass(Value.class); |
| // job.setOutputFormatClass(NullOutputFormat.class); |
| // |
| // job.setNumReduceTasks(0); |
| // |
| // job.waitForCompletion(true); |
| // |
| // return job.isSuccessful() ? 0 : 1; |
| // } |
| // |
| // public static int main(String... args) throws Exception { |
| // Configuration conf = new Configuration(); |
| // conf.set("mapreduce.framework.name", "local"); |
| // conf.set("mapreduce.cluster.local.dir", |
| // new File(System.getProperty("user.dir"), "target/mapreduce-tmp").getAbsolutePath()); |
| // return ToolRunner.run(conf, new CIFTester(), args); |
| // } |
| // } |
| // |
| // @Test |
| // public void test() throws Exception { |
| // client.tableOperations().create(tableName); |
| // BatchWriter bw = client.createBatchWriter(tableName, new BatchWriterConfig()); |
| // |
| // for (Entry<Key,Value> e : data) { |
| // Key k = e.getKey(); |
| // Mutation m = new Mutation(k.getRow()); |
| // m.put(k.getColumnFamily(), k.getColumnQualifier(), |
| // new ColumnVisibility(k.getColumnVisibility()), k.getTimestamp(), e.getValue()); |
| // bw.addMutation(m); |
| // } |
| // bw.close(); |
| // |
| // assertEquals(0, CIFTester.main(tableName, CIFTester.TestMapper.class.getName())); |
| // assertEquals(1, assertionErrors.get(tableName).size()); |
| // } |
| // |
| // @Test |
| // public void testErrorOnNextWithoutClose() throws Exception { |
| // client.tableOperations().create(tableName); |
| // BatchWriter bw = client.createBatchWriter(tableName, new BatchWriterConfig()); |
| // |
| // for (Entry<Key,Value> e : data) { |
| // Key k = e.getKey(); |
| // Mutation m = new Mutation(k.getRow()); |
| // m.put(k.getColumnFamily(), k.getColumnQualifier(), |
| // new ColumnVisibility(k.getColumnVisibility()), k.getTimestamp(), e.getValue()); |
| // bw.addMutation(m); |
| // } |
| // bw.close(); |
| // |
| // assertEquals(1, CIFTester.main(tableName, CIFTester.TestNoClose.class.getName())); |
| // assertEquals(1, assertionErrors.get(tableName).size()); |
| // // this should actually exist, in addition to the dummy entry |
| // assertEquals(2, assertionErrors.get(tableName + "_map_ioexception").size()); |
| // } |
| // |
| // @Test |
| // public void testInfoWithoutChunks() throws Exception { |
| // client.tableOperations().create(tableName); |
| // BatchWriter bw = client.createBatchWriter(tableName, new BatchWriterConfig()); |
| // for (Entry<Key,Value> e : baddata) { |
| // Key k = e.getKey(); |
| // Mutation m = new Mutation(k.getRow()); |
| // m.put(k.getColumnFamily(), k.getColumnQualifier(), |
| // new ColumnVisibility(k.getColumnVisibility()), k.getTimestamp(), e.getValue()); |
| // bw.addMutation(m); |
| // } |
| // bw.close(); |
| // |
| // assertEquals(0, CIFTester.main(tableName, CIFTester.TestBadData.class.getName())); |
| // assertEquals(1, assertionErrors.get(tableName).size()); |
| // } |
| // } |
