diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/RawComparator.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/RawComparator.java index 354dda964e92b1..aeee7a6025f9c5 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/RawComparator.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/RawComparator.java @@ -22,7 +22,6 @@ import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; -import org.apache.hadoop.io.serializer.DeserializerComparator; /** *

@@ -30,7 +29,6 @@ * objects. *

* @param generic type. - * @see DeserializerComparator */ @InterfaceAudience.Public @InterfaceStability.Stable diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/file/tfile/TFile.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/file/tfile/TFile.java index aeacc16a78f9bb..484b024ec5f691 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/file/tfile/TFile.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/file/tfile/TFile.java @@ -48,7 +48,6 @@ import org.apache.hadoop.io.file.tfile.CompareUtils.BytesComparator; import org.apache.hadoop.io.file.tfile.CompareUtils.MemcmpRawComparator; import org.apache.hadoop.io.file.tfile.Utils.Version; -import org.apache.hadoop.io.serializer.JavaSerializationComparator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -268,8 +267,7 @@ private enum State { * Currently, we only support RawComparators that can be * constructed through the default constructor (with no * parameters). Parameterized RawComparators such as - * {@link WritableComparator} or - * {@link JavaSerializationComparator} may not be directly used. + * {@link WritableComparator} may not be directly used. * One should write a wrapper class that inherits from such classes * and use its default constructor to perform proper * initialization. diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/serializer/DeserializerComparator.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/serializer/DeserializerComparator.java deleted file mode 100644 index 29c04f66d43709..00000000000000 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/serializer/DeserializerComparator.java +++ /dev/null @@ -1,75 +0,0 @@ -/** - * 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.hadoop.io.serializer; - -import java.io.IOException; -import java.util.Comparator; - -import org.apache.hadoop.classification.InterfaceAudience; -import org.apache.hadoop.classification.InterfaceStability; -import org.apache.hadoop.io.InputBuffer; -import org.apache.hadoop.io.RawComparator; - -/** - *

- * A {@link RawComparator} that uses a {@link Deserializer} to deserialize - * the objects to be compared so that the standard {@link Comparator} can - * be used to compare them. - *

- *

- * One may optimize compare-intensive operations by using a custom - * implementation of {@link RawComparator} that operates directly - * on byte representations. - *

- * @param generic type. - */ -@InterfaceAudience.LimitedPrivate({"HDFS", "MapReduce"}) -@InterfaceStability.Evolving -public abstract class DeserializerComparator implements RawComparator { - - private InputBuffer buffer = new InputBuffer(); - private Deserializer deserializer; - - private T key1; - private T key2; - - protected DeserializerComparator(Deserializer deserializer) - throws IOException { - - this.deserializer = deserializer; - this.deserializer.open(buffer); - } - - @Override - public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { - try { - - buffer.reset(b1, s1, l1); - key1 = deserializer.deserialize(key1); - - buffer.reset(b2, s2, l2); - key2 = deserializer.deserialize(key2); - - } catch (IOException e) { - throw new RuntimeException(e); - } - return compare(key1, key2); - } - -} diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/serializer/JavaSerialization.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/serializer/JavaSerialization.java deleted file mode 100644 index f08d0008c6eacf..00000000000000 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/serializer/JavaSerialization.java +++ /dev/null @@ -1,117 +0,0 @@ -/** - * 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.hadoop.io.serializer; - -import java.io.IOException; -import java.io.InputStream; -import java.io.ObjectInputStream; -import java.io.ObjectOutputStream; -import java.io.OutputStream; -import java.io.Serializable; -import org.apache.hadoop.classification.InterfaceAudience; -import org.apache.hadoop.classification.InterfaceStability; - -/** - *

- * An experimental {@link Serialization} for Java {@link Serializable} classes. - *

- * @see JavaSerializationComparator - */ -@InterfaceAudience.Public -@InterfaceStability.Unstable -public class JavaSerialization implements Serialization { - - static class JavaSerializationDeserializer - implements Deserializer { - - private ObjectInputStream ois; - - @Override - public void open(InputStream in) throws IOException { - ois = new ObjectInputStream(in) { - @Override protected void readStreamHeader() { - // no header - } - }; - } - - @Override - @SuppressWarnings("unchecked") - public T deserialize(T object) throws IOException { - try { - // ignore passed-in object - return (T) ois.readObject(); - } catch (ClassNotFoundException e) { - throw new IOException(e.toString()); - } - } - - @Override - public void close() throws IOException { - ois.close(); - } - - } - - static class JavaSerializationSerializer - implements Serializer { - - private ObjectOutputStream oos; - - @Override - public void open(OutputStream out) throws IOException { - oos = new ObjectOutputStream(out) { - @Override protected void writeStreamHeader() { - // no header - } - }; - } - - @Override - public void serialize(Serializable object) throws IOException { - oos.reset(); // clear (class) back-references - oos.writeObject(object); - } - - @Override - public void close() throws IOException { - oos.close(); - } - - } - - @Override - @InterfaceAudience.Private - public boolean accept(Class c) { - return Serializable.class.isAssignableFrom(c); - } - - @Override - @InterfaceAudience.Private - public Deserializer getDeserializer(Class c) { - return new JavaSerializationDeserializer(); - } - - @Override - @InterfaceAudience.Private - public Serializer getSerializer(Class c) { - return new JavaSerializationSerializer(); - } - -} diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/serializer/JavaSerializationComparator.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/serializer/JavaSerializationComparator.java deleted file mode 100644 index d53f7ab75c5036..00000000000000 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/serializer/JavaSerializationComparator.java +++ /dev/null @@ -1,53 +0,0 @@ -/** - * 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.hadoop.io.serializer; - -import java.io.IOException; -import java.io.Serializable; - -import org.apache.hadoop.classification.InterfaceAudience; -import org.apache.hadoop.classification.InterfaceStability; -import org.apache.hadoop.io.RawComparator; - -/** - *

- * A {@link RawComparator} that uses a {@link JavaSerialization} - * {@link Deserializer} to deserialize objects that are then compared via - * their {@link Comparable} interfaces. - *

- * @param generic type. - * @see JavaSerialization - */ -@InterfaceAudience.Public -@InterfaceStability.Unstable -public class JavaSerializationComparator> - extends DeserializerComparator { - - @InterfaceAudience.Private - public JavaSerializationComparator() throws IOException { - super(new JavaSerialization.JavaSerializationDeserializer()); - } - - @Override - @InterfaceAudience.Private - public int compare(T o1, T o2) { - return o1.compareTo(o2); - } - -} diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/shell/TestTextCommand.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/shell/TestTextCommand.java index ab4d0b2cc97a01..eba60669f145f7 100644 --- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/shell/TestTextCommand.java +++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/shell/TestTextCommand.java @@ -35,6 +35,7 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.SequenceFile; +import org.apache.hadoop.io.Text; import org.apache.hadoop.test.GenericTestUtils; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; @@ -513,22 +514,20 @@ private static byte[] generateEmptyAvroBinaryData() { private static void createEmptySequenceFile(String fileName, Configuration conf) throws IOException { - conf.set("io.serializations", "org.apache.hadoop.io.serializer.JavaSerialization"); Path path = new Path(fileName); SequenceFile.Writer writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(path), - SequenceFile.Writer.keyClass(String.class), SequenceFile.Writer.valueClass(String.class)); + SequenceFile.Writer.keyClass(Text.class), SequenceFile.Writer.valueClass(Text.class)); writer.close(); } private static void createNonWritableSequenceFile(String fileName, Configuration conf) throws IOException { - conf.set("io.serializations", "org.apache.hadoop.io.serializer.JavaSerialization"); Path path = new Path(fileName); try (SequenceFile.Writer writer = SequenceFile.createWriter(conf, - SequenceFile.Writer.file(path), SequenceFile.Writer.keyClass(String.class), - SequenceFile.Writer.valueClass(String.class))) { - writer.append("Key1", "Value1"); - writer.append("Key2", "Value2"); + SequenceFile.Writer.file(path), SequenceFile.Writer.keyClass(Text.class), + SequenceFile.Writer.valueClass(Text.class))) { + writer.append(new Text("Key1"), new Text("Value1")); + writer.append(new Text("Key2"), new Text("Value2")); } } diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestDefaultStringifier.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestDefaultStringifier.java index 60e12d7bd41c3b..a33ec7d0b6f4a4 100644 --- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestDefaultStringifier.java +++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestDefaultStringifier.java @@ -27,20 +27,20 @@ import org.slf4j.LoggerFactory; import static org.apache.hadoop.test.LambdaTestUtils.intercept; -import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.assertj.core.api.Assertions.assertThat; public class TestDefaultStringifier { - private static Configuration conf = new Configuration(); + private static final Configuration CONF = new Configuration(); private static final Logger LOG = LoggerFactory.getLogger(TestDefaultStringifier.class); - private char[] alphabet = "abcdefghijklmnopqrstuvwxyz".toCharArray(); + private static final char[] ALPHABET = "abcdefghijklmnopqrstuvwxyz".toCharArray(); @Test public void testWithWritable() throws Exception { - conf.set("io.serializations", "org.apache.hadoop.io.serializer.WritableSerialization"); + CONF.set("io.serializations", "org.apache.hadoop.io.serializer.WritableSerialization"); LOG.info("Testing DefaultStringifier with Text"); @@ -52,70 +52,54 @@ public void testWithWritable() throws Exception { StringBuilder builder = new StringBuilder(); int strLen = random.nextInt(40); for(int j=0; j< strLen; j++) { - builder.append(alphabet[random.nextInt(alphabet.length)]); + builder.append(ALPHABET[random.nextInt(ALPHABET.length)]); } Text text = new Text(builder.toString()); - DefaultStringifier stringifier = new DefaultStringifier(conf, Text.class); + DefaultStringifier stringifier = new DefaultStringifier<>(CONF, Text.class); String str = stringifier.toString(text); Text claimedText = stringifier.fromString(str); - LOG.info("Object: " + text); - LOG.info("String representation of the object: " + str); - assertEquals(text, claimedText); + LOG.info("Object: {}", text); + LOG.info("String representation of the object: {}", str); + assertThat(claimedText).isEqualTo(text); } } - @Test - public void testWithJavaSerialization() throws Exception { - conf.set("io.serializations", "org.apache.hadoop.io.serializer.JavaSerialization"); - - LOG.info("Testing DefaultStringifier with Serializable Integer"); - - //Integer implements Serializable - Integer testInt = Integer.valueOf(42); - DefaultStringifier stringifier = new DefaultStringifier(conf, Integer.class); - - String str = stringifier.toString(testInt); - Integer claimedInt = stringifier.fromString(str); - LOG.info("String representation of the object: " + str); - - assertEquals(testInt, claimedInt); - } - @Test public void testStoreLoad() throws IOException { LOG.info("Testing DefaultStringifier#store() and #load()"); - conf.set("io.serializations", "org.apache.hadoop.io.serializer.WritableSerialization"); + CONF.set("io.serializations", "org.apache.hadoop.io.serializer.WritableSerialization"); Text text = new Text("uninteresting test string"); String keyName = "test.defaultstringifier.key1"; - DefaultStringifier.store(conf,text, keyName); - - Text claimedText = DefaultStringifier.load(conf, keyName, Text.class); - assertEquals(text, claimedText, - "DefaultStringifier#load() or #store() might be flawed"); + DefaultStringifier.store(CONF, text, keyName); + Text claimedText = DefaultStringifier.load(CONF, keyName, Text.class); + assertThat(claimedText) + .describedAs("DefaultStringifier round trip") + .isEqualTo(text); } @Test public void testStoreLoadArray() throws Exception { LOG.info("Testing DefaultStringifier#storeArray() and #loadArray()"); - conf.set("io.serializations", "org.apache.hadoop.io.serializer.JavaSerialization"); + CONF.set("io.serializations", "org.apache.hadoop.io.serializer.WritableSerialization"); String keyName = "test.defaultstringifier.key2"; - Integer[] array = new Integer[] {1,2,3,4,5}; + IntWritable[] array = new IntWritable[] { + new IntWritable(1), new IntWritable(2), new IntWritable(3), + new IntWritable(4), new IntWritable(5)}; intercept(IndexOutOfBoundsException.class, () -> - DefaultStringifier.storeArray(conf, new Integer[] {}, keyName)); - DefaultStringifier.storeArray(conf, array, keyName); + DefaultStringifier.storeArray(CONF, new IntWritable[] {}, keyName)); + DefaultStringifier.storeArray(CONF, array, keyName); - Integer[] claimedArray = DefaultStringifier.loadArray(conf, keyName, Integer.class); - for (int i = 0; i < array.length; i++) { - assertEquals(array[i], claimedArray[i], "two arrays are not equal"); - } + IntWritable[] claimedArray = + DefaultStringifier.loadArray(CONF, keyName, IntWritable.class); + assertThat(claimedArray).isEqualTo(array); } diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestSequenceFileAppend.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestSequenceFileAppend.java index fe2ea36fc6d25c..46b1ac0f322f91 100644 --- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestSequenceFileAppend.java +++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestSequenceFileAppend.java @@ -18,8 +18,8 @@ package org.apache.hadoop.io; +import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.fail; import java.io.IOException; @@ -33,7 +33,6 @@ import org.apache.hadoop.io.SequenceFile.Writer.Option; import org.apache.hadoop.io.compress.DefaultCodec; import org.apache.hadoop.io.compress.GzipCodec; -import org.apache.hadoop.io.serializer.JavaSerializationComparator; import org.apache.hadoop.test.GenericTestUtils; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; @@ -51,7 +50,7 @@ public class TestSequenceFileAppend { public static void setUp() throws Exception { conf = new Configuration(); conf.set("io.serializations", - "org.apache.hadoop.io.serializer.JavaSerialization"); + "org.apache.hadoop.io.serializer.WritableSerialization"); conf.set("fs.file.impl", "org.apache.hadoop.fs.RawLocalFileSystem"); fs = FileSystem.get(conf); } @@ -78,11 +77,11 @@ public void testAppend() throws Exception { Writer writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), metadataOption); + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), metadataOption); - writer.append(1L, "one"); - writer.append(2L, "two"); + writer.append(new LongWritable(1L), new Text("one")); + writer.append(new LongWritable(2L), new Text("two")); writer.close(); verify2Values(file); @@ -90,15 +89,15 @@ public void testAppend() throws Exception { metadata.set(key1, value2); writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), SequenceFile.Writer.appendIfExists(true), metadataOption); // Verify the Meta data is not changed assertEquals(value1, writer.metadata.get(key1)); - writer.append(3L, "three"); - writer.append(4L, "four"); + writer.append(new LongWritable(3L), new Text("three")); + writer.append(new LongWritable(4L), new Text("four")); writer.close(); @@ -115,8 +114,8 @@ public void testAppend() throws Exception { new GzipCodec()); writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), SequenceFile.Writer.appendIfExists(true), wrongCompressOption); writer.close(); fail("Expected IllegalArgumentException for compression options"); @@ -129,8 +128,8 @@ public void testAppend() throws Exception { new DefaultCodec()); writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), SequenceFile.Writer.appendIfExists(true), wrongCompressOption); writer.close(); fail("Expected IllegalArgumentException for compression options"); @@ -153,22 +152,22 @@ public void testAppendRecordCompression() throws Exception { new GzipCodec()); Writer writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), compressOption); + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), compressOption); - writer.append(1L, "one"); - writer.append(2L, "two"); + writer.append(new LongWritable(1L), new Text("one")); + writer.append(new LongWritable(2L), new Text("two")); writer.close(); verify2Values(file); writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), SequenceFile.Writer.appendIfExists(true), compressOption); - writer.append(3L, "three"); - writer.append(4L, "four"); + writer.append(new LongWritable(3L), new Text("three")); + writer.append(new LongWritable(4L), new Text("four")); writer.close(); verifyAll4Values(file); @@ -188,22 +187,22 @@ public void testAppendBlockCompression() throws Exception { new GzipCodec()); Writer writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), compressOption); + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), compressOption); - writer.append(1L, "one"); - writer.append(2L, "two"); + writer.append(new LongWritable(1L), new Text("one")); + writer.append(new LongWritable(2L), new Text("two")); writer.close(); verify2Values(file); writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), SequenceFile.Writer.appendIfExists(true), compressOption); - writer.append(3L, "three"); - writer.append(4L, "four"); + writer.append(new LongWritable(3L), new Text("three")); + writer.append(new LongWritable(4L), new Text("four")); writer.close(); verifyAll4Values(file); @@ -211,8 +210,8 @@ public void testAppendBlockCompression() throws Exception { // Verify failure if the compression details are different or not Provided try { writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), SequenceFile.Writer.appendIfExists(true)); writer.close(); fail("Expected IllegalArgumentException for compression options"); @@ -226,8 +225,8 @@ public void testAppendBlockCompression() throws Exception { new GzipCodec()); writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), SequenceFile.Writer.appendIfExists(true), wrongCompressOption); writer.close(); fail("Expected IllegalArgumentException for compression options"); @@ -240,8 +239,8 @@ public void testAppendBlockCompression() throws Exception { new DefaultCodec()); writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), SequenceFile.Writer.appendIfExists(true), wrongCompressOption); writer.close(); fail("Expected IllegalArgumentException for compression options"); @@ -261,22 +260,22 @@ public void testAppendNoneCompression() throws Exception { Option compressOption = Writer.compression(CompressionType.NONE); Writer writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), compressOption); + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), compressOption); - writer.append(1L, "one"); - writer.append(2L, "two"); + writer.append(new LongWritable(1L), new Text("one")); + writer.append(new LongWritable(2L), new Text("two")); writer.close(); verify2Values(file); writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), SequenceFile.Writer.appendIfExists(true), compressOption); - writer.append(3L, "three"); - writer.append(4L, "four"); + writer.append(new LongWritable(3L), new Text("three")); + writer.append(new LongWritable(4L), new Text("four")); writer.close(); verifyAll4Values(file); @@ -284,8 +283,8 @@ public void testAppendNoneCompression() throws Exception { // Verify failure if the compression details are different or not Provided try { writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), SequenceFile.Writer.appendIfExists(true)); writer.close(); fail("Expected IllegalArgumentException for compression options"); @@ -299,8 +298,8 @@ public void testAppendNoneCompression() throws Exception { Writer.compression(CompressionType.RECORD, new GzipCodec()); writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), SequenceFile.Writer.appendIfExists(true), wrongCompressOption); writer.close(); fail("Expected IllegalArgumentException for compression options"); @@ -313,77 +312,34 @@ public void testAppendNoneCompression() throws Exception { Writer.compression(CompressionType.NONE, new DefaultCodec()); writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), + SequenceFile.Writer.keyClass(LongWritable.class), + SequenceFile.Writer.valueClass(Text.class), SequenceFile.Writer.appendIfExists(true), noneWithCodec); writer.close(); fs.deleteOnExit(file); } - @Test - @Timeout(value = 30) - public void testAppendSort() throws Exception { - GenericTestUtils.assumeInNativeProfile(); - - Path file = new Path(ROOT_PATH, "testseqappendSort.seq"); - fs.delete(file, true); - - Path sortedFile = new Path(ROOT_PATH, "testseqappendSort.seq.sort"); - fs.delete(sortedFile, true); - - SequenceFile.Sorter sorter = new SequenceFile.Sorter(fs, - new JavaSerializationComparator(), Long.class, String.class, conf); - - Option compressOption = Writer.compression(CompressionType.BLOCK, - new GzipCodec()); - Writer writer = SequenceFile.createWriter(conf, - SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), compressOption); - - writer.append(2L, "two"); - writer.append(1L, "one"); - - writer.close(); - - writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(file), - SequenceFile.Writer.keyClass(Long.class), - SequenceFile.Writer.valueClass(String.class), - SequenceFile.Writer.appendIfExists(true), compressOption); - - writer.append(4L, "four"); - writer.append(3L, "three"); - writer.close(); - - // Sort file after append - sorter.sort(file, sortedFile); - verifyAll4Values(sortedFile); - - fs.deleteOnExit(file); - fs.deleteOnExit(sortedFile); - } - private void verify2Values(Path file) throws IOException { Reader reader = new Reader(conf, Reader.file(file)); - assertEquals(1L, reader.next((Object) null)); - assertEquals("one", reader.getCurrentValue((Object) null)); - assertEquals(2L, reader.next((Object) null)); - assertEquals("two", reader.getCurrentValue((Object) null)); - assertNull(reader.next((Object) null)); + assertThat(reader.next((Object) null)).isEqualTo(new LongWritable(1L)); + assertThat(reader.getCurrentValue((Object) null)).isEqualTo(new Text("one")); + assertThat(reader.next((Object) null)).isEqualTo(new LongWritable(2L)); + assertThat(reader.getCurrentValue((Object) null)).isEqualTo(new Text("two")); + assertThat(reader.next((Object) null)).isNull(); reader.close(); } private void verifyAll4Values(Path file) throws IOException { Reader reader = new Reader(conf, Reader.file(file)); - assertEquals(1L, reader.next((Object) null)); - assertEquals("one", reader.getCurrentValue((Object) null)); - assertEquals(2L, reader.next((Object) null)); - assertEquals("two", reader.getCurrentValue((Object) null)); - assertEquals(3L, reader.next((Object) null)); - assertEquals("three", reader.getCurrentValue((Object) null)); - assertEquals(4L, reader.next((Object) null)); - assertEquals("four", reader.getCurrentValue((Object) null)); - assertNull(reader.next((Object) null)); + assertThat(reader.next((Object) null)).isEqualTo(new LongWritable(1L)); + assertThat(reader.getCurrentValue((Object) null)).isEqualTo(new Text("one")); + assertThat(reader.next((Object) null)).isEqualTo(new LongWritable(2L)); + assertThat(reader.getCurrentValue((Object) null)).isEqualTo(new Text("two")); + assertThat(reader.next((Object) null)).isEqualTo(new LongWritable(3L)); + assertThat(reader.getCurrentValue((Object) null)).isEqualTo(new Text("three")); + assertThat(reader.next((Object) null)).isEqualTo(new LongWritable(4L)); + assertThat(reader.getCurrentValue((Object) null)).isEqualTo(new Text("four")); + assertThat(reader.next((Object) null)).isNull(); reader.close(); } } diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestSequenceFileSerialization.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestSequenceFileSerialization.java deleted file mode 100644 index 90fe5ab589a0b2..00000000000000 --- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestSequenceFileSerialization.java +++ /dev/null @@ -1,74 +0,0 @@ -/** - * 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.hadoop.io; - - -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.io.SequenceFile.Reader; -import org.apache.hadoop.io.SequenceFile.Writer; -import org.apache.hadoop.test.GenericTestUtils; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; - -import static org.junit.jupiter.api.Assertions.assertNull; -import static org.junit.jupiter.api.Assertions.assertEquals; - -public class TestSequenceFileSerialization { - private Configuration conf; - private FileSystem fs; - - @BeforeEach - public void setUp() throws Exception { - conf = new Configuration(); - conf.set("io.serializations", - "org.apache.hadoop.io.serializer.JavaSerialization"); - fs = FileSystem.getLocal(conf); - } - - @AfterEach - public void tearDown() throws Exception { - fs.close(); - } - - @Test - public void testJavaSerialization() throws Exception { - Path file = new Path(GenericTestUtils.getTempPath("testseqser.seq")); - - fs.delete(file, true); - Writer writer = SequenceFile.createWriter(fs, conf, file, Long.class, - String.class); - - writer.append(1L, "one"); - writer.append(2L, "two"); - - writer.close(); - - Reader reader = new Reader(fs, file, conf); - assertEquals(1L, reader.next((Object) null)); - assertEquals("one", reader.getCurrentValue((Object) null)); - assertEquals(2L, reader.next((Object) null)); - assertEquals("two", reader.getCurrentValue((Object) null)); - assertNull(reader.next((Object) null)); - reader.close(); - - } -} diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/serializer/TestWritableSerialization.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/serializer/TestWritableSerialization.java index beea0cf4bb76fe..cdc63011c632e0 100644 --- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/serializer/TestWritableSerialization.java +++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/serializer/TestWritableSerialization.java @@ -18,10 +18,6 @@ package org.apache.hadoop.io.serializer; -import java.io.Serializable; - -import org.apache.hadoop.io.DataInputBuffer; -import org.apache.hadoop.io.DataOutputBuffer; import static org.apache.hadoop.io.TestGenericWritable.CONF_TEST_KEY; import static org.apache.hadoop.io.TestGenericWritable.CONF_TEST_VALUE; @@ -29,7 +25,6 @@ import org.apache.hadoop.io.Text; import org.apache.hadoop.io.TestGenericWritable.Baz; import org.apache.hadoop.io.TestGenericWritable.FooGenericWritable; -import org.apache.hadoop.io.WritableComparator; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -63,41 +58,4 @@ public void testWritableConfigurable() throws Exception { assertNotNull(result.getConf()); } - @Test - @SuppressWarnings({"rawtypes", "unchecked"}) - public void testWritableComparatorJavaSerialization() throws Exception { - Serialization ser = new JavaSerialization(); - - Serializer serializer = ser.getSerializer(TestWC.class); - DataOutputBuffer dob = new DataOutputBuffer(); - serializer.open(dob); - TestWC orig = new TestWC(0); - serializer.serialize(orig); - serializer.close(); - - Deserializer deserializer = ser.getDeserializer(TestWC.class); - DataInputBuffer dib = new DataInputBuffer(); - dib.reset(dob.getData(), 0, dob.getLength()); - deserializer.open(dib); - TestWC deser = deserializer.deserialize(null); - deserializer.close(); - assertEquals(orig, deser); - } - - static class TestWC extends WritableComparator implements Serializable { - static final long serialVersionUID = 0x4344; - final int val; - TestWC() { this(7); } - TestWC(int val) { this.val = val; } - @Override - public boolean equals(Object o) { - if (o instanceof TestWC) { - return ((TestWC)o).val == val; - } - return false; - } - @Override - public int hashCode() { return val; } - } - } diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestJavaSerialization.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestJavaSerialization.java deleted file mode 100644 index 532020d98981c7..00000000000000 --- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestJavaSerialization.java +++ /dev/null @@ -1,192 +0,0 @@ -/** - * 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.hadoop.mapred; - -import java.io.File; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.io.OutputStreamWriter; -import java.io.Writer; -import java.nio.charset.StandardCharsets; -import java.util.Iterator; -import java.util.StringTokenizer; - -import org.apache.commons.io.FileUtils; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.FileUtil; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.io.LongWritable; -import org.apache.hadoop.io.Text; -import org.apache.hadoop.io.serializer.JavaSerializationComparator; -import org.apache.hadoop.mapreduce.MRConfig; -import org.junit.jupiter.api.Test; -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.junit.jupiter.api.Assertions.assertEquals; - -public class TestJavaSerialization { - - private static String TEST_ROOT_DIR = - new File(System.getProperty("test.build.data", "/tmp")).toURI() - .toString().replace(' ', '+'); - - private final Path INPUT_DIR = new Path(TEST_ROOT_DIR + "/input"); - private final Path OUTPUT_DIR = new Path(TEST_ROOT_DIR + "/out"); - private final Path INPUT_FILE = new Path(INPUT_DIR , "inp"); - - static class WordCountMapper extends MapReduceBase implements - Mapper { - - public void map(LongWritable key, Text value, - OutputCollector output, Reporter reporter) - throws IOException { - StringTokenizer st = new StringTokenizer(value.toString()); - while (st.hasMoreTokens()) { - String token = st.nextToken(); - assertTrue(token.equals("a") || token.equals("b"), - "Invalid token; expected 'a' or 'b', got " + token); - output.collect(token, 1L); - } - } - - } - - static class SumReducer extends MapReduceBase implements - Reducer { - - public void reduce(K key, Iterator values, - OutputCollector output, Reporter reporter) - throws IOException { - - long sum = 0; - while (values.hasNext()) { - sum += values.next(); - } - output.collect(key, sum); - } - - } - - private void cleanAndCreateInput(FileSystem fs) throws IOException { - fs.delete(INPUT_DIR, true); - fs.delete(OUTPUT_DIR, true); - - OutputStream os = fs.create(INPUT_FILE); - - Writer wr = new OutputStreamWriter(os); - wr.write("b a\n"); - wr.close(); - } - - @SuppressWarnings("deprecation") - @Test - public void testMapReduceJob() throws Exception { - - JobConf conf = new JobConf(TestJavaSerialization.class); - conf.setJobName("JavaSerialization"); - - FileSystem fs = FileSystem.get(conf); - cleanAndCreateInput(fs); - - conf.set("io.serializations", - "org.apache.hadoop.io.serializer.JavaSerialization," + - "org.apache.hadoop.io.serializer.WritableSerialization"); - - conf.setInputFormat(TextInputFormat.class); - - conf.setOutputKeyClass(String.class); - conf.setOutputValueClass(Long.class); - conf.setOutputKeyComparatorClass(JavaSerializationComparator.class); - - conf.setMapperClass(WordCountMapper.class); - conf.setReducerClass(SumReducer.class); - - conf.set(MRConfig.FRAMEWORK_NAME, MRConfig.LOCAL_FRAMEWORK_NAME); - - FileInputFormat.setInputPaths(conf, INPUT_DIR); - - FileOutputFormat.setOutputPath(conf, OUTPUT_DIR); - - String inputFileContents = - FileUtils.readFileToString(new File(INPUT_FILE.toUri().getPath())); - assertTrue(inputFileContents.equals("b a\n"), - "Input file contents not as expected; contents are '" - + inputFileContents + "', expected \"b a\n\" "); - - JobClient.runJob(conf); - - Path[] outputFiles = - FileUtil.stat2Paths(fs.listStatus(OUTPUT_DIR, - new Utils.OutputFileUtils.OutputFilesFilter())); - assertEquals(1, outputFiles.length); - try (InputStream is = fs.open(outputFiles[0])) { - String reduceOutput = org.apache.commons.io.IOUtils.toString(is, StandardCharsets.UTF_8); - String[] lines = reduceOutput.split("\n"); - assertEquals("a\t1", lines[0], - "Unexpected output; received output '" + reduceOutput + "'"); - assertEquals("b\t1", lines[1], - "Unexpected output; received output '" + reduceOutput + "'"); - assertEquals(2, lines.length, - "Reduce output has extra lines; output is '" + reduceOutput + "'"); - } - } - - /** - * HADOOP-4466: - * This test verifies the JavSerialization impl can write to - * SequenceFiles. by virtue other SequenceFileOutputFormat is not - * coupled to Writable types, if so, the job will fail. - * - */ - @Test - public void testWriteToSequencefile() throws Exception { - JobConf conf = new JobConf(TestJavaSerialization.class); - conf.setJobName("JavaSerialization"); - - FileSystem fs = FileSystem.get(conf); - cleanAndCreateInput(fs); - - conf.set("io.serializations", - "org.apache.hadoop.io.serializer.JavaSerialization," + - "org.apache.hadoop.io.serializer.WritableSerialization"); - - conf.setInputFormat(TextInputFormat.class); - // test we can write to sequence files - conf.setOutputFormat(SequenceFileOutputFormat.class); - conf.setOutputKeyClass(String.class); - conf.setOutputValueClass(Long.class); - conf.setOutputKeyComparatorClass(JavaSerializationComparator.class); - - conf.setMapperClass(WordCountMapper.class); - conf.setReducerClass(SumReducer.class); - - conf.set(MRConfig.FRAMEWORK_NAME, MRConfig.LOCAL_FRAMEWORK_NAME); - - FileInputFormat.setInputPaths(conf, INPUT_DIR); - - FileOutputFormat.setOutputPath(conf, OUTPUT_DIR); - - JobClient.runJob(conf); - - Path[] outputFiles = FileUtil.stat2Paths( - fs.listStatus(OUTPUT_DIR, - new Utils.OutputFileUtils.OutputFilesFilter())); - assertEquals(1, outputFiles.length); - } - -} diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/lib/TestMultipleOutputs.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/lib/TestMultipleOutputs.java index 41cd160025428c..2452a3eab972eb 100644 --- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/lib/TestMultipleOutputs.java +++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/lib/TestMultipleOutputs.java @@ -23,7 +23,6 @@ import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.SequenceFile; import org.apache.hadoop.io.Text; -import org.apache.hadoop.io.serializer.JavaSerializationComparator; import org.apache.hadoop.mapred.Counters; import org.apache.hadoop.mapred.FileInputFormat; import org.apache.hadoop.mapred.FileOutputFormat; @@ -68,13 +67,11 @@ public TestMultipleOutputs() throws IOException { @Test public void testWithoutCounters() throws Exception { _testMultipleOutputs(false); - _testMOWithJavaSerialization(false); } @Test public void testWithCounters() throws Exception { _testMultipleOutputs(true); - _testMOWithJavaSerialization(true); } @SuppressWarnings("unchecked") @@ -130,94 +127,6 @@ public void tearDown() throws Exception { super.tearDown(); } - protected void _testMOWithJavaSerialization(boolean withCounters) throws Exception { - Path inDir = getDir(IN_DIR); - Path outDir = getDir(OUT_DIR); - - JobConf conf = createJobConf(); - FileSystem fs = FileSystem.get(conf); - - DataOutputStream file = fs.create(new Path(inDir, "part-0")); - file.writeBytes("a\nb\n\nc\nd\ne"); - file.close(); - - fs.delete(inDir, true); - fs.delete(outDir, true); - - file = fs.create(new Path(inDir, "part-1")); - file.writeBytes("a\nb\n\nc\nd\ne"); - file.close(); - - conf.setJobName("mo"); - - conf.set("io.serializations", - "org.apache.hadoop.io.serializer.JavaSerialization," + - "org.apache.hadoop.io.serializer.WritableSerialization"); - - conf.setInputFormat(TextInputFormat.class); - - conf.setMapOutputKeyClass(Long.class); - conf.setMapOutputValueClass(String.class); - conf.setOutputKeyComparatorClass(JavaSerializationComparator.class); - - conf.setOutputKeyClass(Long.class); - conf.setOutputValueClass(String.class); - - conf.setOutputFormat(TextOutputFormat.class); - - MultipleOutputs.addNamedOutput(conf, "text", TextOutputFormat.class, - Long.class, String.class); - - MultipleOutputs.setCountersEnabled(conf, withCounters); - - conf.setMapperClass(MOJavaSerDeMap.class); - conf.setReducerClass(MOJavaSerDeReduce.class); - - FileInputFormat.setInputPaths(conf, inDir); - FileOutputFormat.setOutputPath(conf, outDir); - - JobClient jc = new JobClient(conf); - RunningJob job = jc.submitJob(conf); - while (!job.isComplete()) { - Thread.sleep(100); - } - - // assert number of named output part files - int namedOutputCount = 0; - FileStatus[] statuses = fs.listStatus(outDir); - for (FileStatus status : statuses) { - if (status.getPath().getName().equals("text-m-00000") || - status.getPath().getName().equals("text-r-00000")) { - namedOutputCount++; - } - } - assertEquals(2, namedOutputCount); - - // assert TextOutputFormat files correctness - BufferedReader reader = new BufferedReader( - new InputStreamReader(fs.open( - new Path(FileOutputFormat.getOutputPath(conf), "text-r-00000")))); - int count = 0; - String line = reader.readLine(); - while (line != null) { - assertTrue(line.endsWith("text")); - line = reader.readLine(); - count++; - } - reader.close(); - assertNotEquals(0, count); - - Counters.Group counters = - job.getCounters().getGroup(MultipleOutputs.class.getName()); - if (!withCounters) { - assertEquals(0, counters.size()); - } - else { - assertEquals(1, counters.size()); - assertEquals(2, counters.getCounter("text")); - } - } - protected void _testMultipleOutputs(boolean withCounters) throws Exception { Path inDir = getDir(IN_DIR); Path outDir = getDir(OUT_DIR); @@ -392,61 +301,5 @@ public void close() throws IOException { mos.close(); } } - - @SuppressWarnings({"unchecked"}) - public static class MOJavaSerDeMap implements Mapper { - - private MultipleOutputs mos; - - public void configure(JobConf conf) { - mos = new MultipleOutputs(conf); - } - - public void map(LongWritable key, Text value, - OutputCollector output, - Reporter reporter) - throws IOException { - if (!value.toString().equals("a")) { - output.collect(key.get(), value.toString()); - } else { - mos.getCollector("text", reporter).collect(key, "text"); - } - } - - public void close() throws IOException { - mos.close(); - } - } - - @SuppressWarnings({"unchecked"}) - public static class MOJavaSerDeReduce implements Reducer { - - private MultipleOutputs mos; - - public void configure(JobConf conf) { - mos = new MultipleOutputs(conf); - } - - public void reduce(Long key, Iterator values, - OutputCollector output, - Reporter reporter) - throws IOException { - while (values.hasNext()) { - String value = values.next(); - if (!value.equals("b")) { - output.collect(key, value); - } else { - mos.getCollector("text", reporter).collect(key, "text"); - } - } - } - - public void close() throws IOException { - mos.close(); - } - } - } diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/lib/output/TestMRMultipleOutputs.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/lib/output/TestMRMultipleOutputs.java index ac8c015a219f1d..61ea77eb9bfa6e 100644 --- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/lib/output/TestMRMultipleOutputs.java +++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/lib/output/TestMRMultipleOutputs.java @@ -25,7 +25,6 @@ import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.SequenceFile; import org.apache.hadoop.io.Text; -import org.apache.hadoop.io.serializer.JavaSerializationComparator; import org.apache.hadoop.mapred.HadoopTestCase; import org.apache.hadoop.mapreduce.CounterGroup; import org.apache.hadoop.mapreduce.Job; @@ -61,13 +60,11 @@ public TestMRMultipleOutputs() throws IOException { @Test public void testWithoutCounters() throws Exception { _testMultipleOutputs(false); - _testMOWithJavaSerialization(false); } @Test public void testWithCounters() throws Exception { _testMultipleOutputs(true); - _testMOWithJavaSerialization(true); } @SuppressWarnings("unchecked") @@ -110,84 +107,6 @@ public void tearDown() throws Exception { super.tearDown(); } - protected void _testMOWithJavaSerialization(boolean withCounters) throws Exception { - String input = "a\nb\nc\nd\ne\nc\nd\ne"; - - Configuration conf = createJobConf(); - conf.set("io.serializations", - "org.apache.hadoop.io.serializer.JavaSerialization," + - "org.apache.hadoop.io.serializer.WritableSerialization"); - - Job job = MapReduceTestUtil.createJob(conf, IN_DIR, OUT_DIR, 2, 1, input); - - job.setJobName("mo"); - MultipleOutputs.addNamedOutput(job, TEXT, TextOutputFormat.class, - Long.class, String.class); - - MultipleOutputs.setCountersEnabled(job, withCounters); - - job.setSortComparatorClass(JavaSerializationComparator.class); - - job.setMapOutputKeyClass(Long.class); - job.setMapOutputValueClass(String.class); - - job.setOutputKeyClass(Long.class); - job.setOutputValueClass(String.class); - - job.setMapperClass(MOJavaSerDeMap.class); - job.setReducerClass(MOJavaSerDeReduce.class); - - job.waitForCompletion(true); - - // assert number of named output part files - int namedOutputCount = 0; - int valueBasedOutputCount = 0; - FileSystem fs = OUT_DIR.getFileSystem(conf); - FileStatus[] statuses = fs.listStatus(OUT_DIR); - for (FileStatus status : statuses) { - String fileName = status.getPath().getName(); - if (fileName.equals("text-m-00000") || - fileName.equals("text-m-00001") || - fileName.equals("text-r-00000")) { - namedOutputCount++; - } else if (fileName.equals("a-r-00000") || - fileName.equals("b-r-00000") || - fileName.equals("c-r-00000") || - fileName.equals("d-r-00000") || - fileName.equals("e-r-00000")) { - valueBasedOutputCount++; - } - } - assertEquals(3, namedOutputCount); - assertEquals(5, valueBasedOutputCount); - - // assert TextOutputFormat files correctness - BufferedReader reader = new BufferedReader( - new InputStreamReader(fs.open( - new Path(FileOutputFormat.getOutputPath(job), "text-r-00000")))); - int count = 0; - String line = reader.readLine(); - while (line != null) { - assertTrue(line.endsWith(TEXT)); - line = reader.readLine(); - count++; - } - reader.close(); - assertFalse(count == 0); - - if (withCounters) { - CounterGroup counters = - job.getCounters().getGroup(MultipleOutputs.class.getName()); - assertEquals(6, counters.size()); - assertEquals(4, counters.findCounter(TEXT).getValue()); - assertEquals(2, counters.findCounter("a").getValue()); - assertEquals(2, counters.findCounter("b").getValue()); - assertEquals(4, counters.findCounter("c").getValue()); - assertEquals(4, counters.findCounter("d").getValue()); - assertEquals(4, counters.findCounter("e").getValue()); - } - } - protected void _testMultipleOutputs(boolean withCounters) throws Exception { String input = "a\nb\nc\nd\ne\nc\nd\ne"; @@ -337,57 +256,7 @@ public void reduce(LongWritable key, Iterable values, } } - public void cleanup(Context context) - throws IOException, InterruptedException { - mos.close(); - } - } - - public static class MOJavaSerDeMap extends Mapper { - - private MultipleOutputs mos; - - public void setup(Context context) { - mos = new MultipleOutputs(context); - } - - public void map(LongWritable key, Text value, Context context) - throws IOException, InterruptedException { - context.write(key.get(), value.toString()); - if (value.toString().equals("a")) { - mos.write(TEXT, key.get(), TEXT); - } - } - - public void cleanup(Context context) - throws IOException, InterruptedException { - mos.close(); - } - } - - public static class MOJavaSerDeReduce extends Reducer { - - private MultipleOutputs mos; - - public void setup(Context context) { - mos = new MultipleOutputs(context); - } - - public void reduce(Long key, Iterable values, - Context context) throws IOException, InterruptedException { - for (String value : values) { - mos.write(key, value, value.toString()); - if (!value.toString().equals("b")) { - context.write(key, value); - } else { - mos.write(TEXT, key, new Text(TEXT)); - } - } - } - - public void cleanup(Context context) + public void cleanup(Context context) throws IOException, InterruptedException { mos.close(); } diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/lib/partition/TestTotalOrderPartitioner.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/lib/partition/TestTotalOrderPartitioner.java index 624909d1286372..e1eeefc84e99e2 100644 --- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/lib/partition/TestTotalOrderPartitioner.java +++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/lib/partition/TestTotalOrderPartitioner.java @@ -21,10 +21,8 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; -import java.util.Comparator; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.CommonConfigurationKeys; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; @@ -34,13 +32,10 @@ import org.apache.hadoop.io.WritableComparator; import org.apache.hadoop.io.WritableUtils; import org.apache.hadoop.io.SequenceFile.CompressionType; -import org.apache.hadoop.io.serializer.JavaSerialization; -import org.apache.hadoop.io.serializer.JavaSerializationComparator; -import org.apache.hadoop.io.serializer.WritableSerialization; import org.apache.hadoop.mapreduce.MRJobConfig; -import org.junit.jupiter.api.Test; -import static org.junit.jupiter.api.Assertions.assertEquals; +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; public class TestTotalOrderPartitioner { @@ -57,19 +52,6 @@ public class TestTotalOrderPartitioner { new Text("yak"), // 9 }; - private static final String[] splitJavaStrings = new String[] { - // -inf // 0 - new String("aabbb"), // 1 - new String("babbb"), // 2 - new String("daddd"), // 3 - new String("dddee"), // 4 - new String("ddhee"), // 5 - new String("dingo"), // 6 - new String("hijjj"), // 7 - new String("n"), // 8 - new String("yak"), // 9 - }; - static class Check { T data; int part; @@ -95,23 +77,6 @@ static class Check { testStrings.add(new Check(new Text("hi"), 6)); }; - private static final ArrayList> testJavaStrings = - new ArrayList>(); - static { - testJavaStrings.add(new Check(new String("aaaaa"), 0)); - testJavaStrings.add(new Check(new String("aaabb"), 0)); - testJavaStrings.add(new Check(new String("aabbb"), 1)); - testJavaStrings.add(new Check(new String("aaaaa"), 0)); - testJavaStrings.add(new Check(new String("babbb"), 2)); - testJavaStrings.add(new Check(new String("baabb"), 1)); - testJavaStrings.add(new Check(new String("yai"), 8)); - testJavaStrings.add(new Check(new String("yak"), 9)); - testJavaStrings.add(new Check(new String("z"), 9)); - testJavaStrings.add(new Check(new String("ddngo"), 5)); - testJavaStrings.add(new Check(new String("hi"), 6)); - }; - - private static Path writePartitionFile( String testname, Configuration conf, T[] splits) throws IOException { final FileSystem fs = FileSystem.getLocal(conf); @@ -122,51 +87,19 @@ private static Path writePartitionFile( Path p = new Path(testdir, testname + "/_partition.lst"); TotalOrderPartitioner.setPartitionFile(conf, p); conf.setInt(MRJobConfig.NUM_REDUCES, splits.length + 1); - SequenceFile.Writer w = null; - try { - w = SequenceFile.createWriter( - conf, - SequenceFile.Writer.file(p), - SequenceFile.Writer.keyClass(splits[0].getClass()), - SequenceFile.Writer.valueClass(NullWritable.class), - SequenceFile.Writer.compression(CompressionType.NONE)); - for (int i = 0; i < splits.length; ++i) { - w.append(splits[i], NullWritable.get()); + try (SequenceFile.Writer w = SequenceFile.createWriter( + conf, + SequenceFile.Writer.file(p), + SequenceFile.Writer.keyClass(splits[0].getClass()), + SequenceFile.Writer.valueClass(NullWritable.class), + SequenceFile.Writer.compression(CompressionType.NONE))) { + for (T split : splits) { + w.append(split, NullWritable.get()); } - } finally { - if (null != w) - w.close(); } return p; } - @Test - public void testTotalOrderWithCustomSerialization() throws Exception { - TotalOrderPartitioner partitioner = - new TotalOrderPartitioner(); - Configuration conf = new Configuration(); - conf.setStrings(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, - JavaSerialization.class.getName(), - WritableSerialization.class.getName()); - conf.setClass(MRJobConfig.KEY_COMPARATOR, - JavaSerializationComparator.class, - Comparator.class); - Path p = TestTotalOrderPartitioner.writePartitionFile( - "totalordercustomserialization", conf, splitJavaStrings); - conf.setClass(MRJobConfig.MAP_OUTPUT_KEY_CLASS, String.class, Object.class); - try { - partitioner.setConf(conf); - NullWritable nw = NullWritable.get(); - for (Check chk : testJavaStrings) { - assertEquals(chk.part, - partitioner.getPartition(chk.data, nw, splitJavaStrings.length + 1), - chk.data.toString()); - } - } finally { - p.getFileSystem(conf).delete(p, true); - } - } - @Test public void testTotalOrderMemCmp() throws Exception { TotalOrderPartitioner partitioner = @@ -179,8 +112,9 @@ public void testTotalOrderMemCmp() throws Exception { partitioner.setConf(conf); NullWritable nw = NullWritable.get(); for (Check chk : testStrings) { - assertEquals(chk.part, - partitioner.getPartition(chk.data, nw, splitStrings.length + 1), chk.data.toString()); + Assertions.assertThat(partitioner.getPartition(chk.data, nw, splitStrings.length + 1)) + .describedAs("Expected %s", chk.data) + .isEqualTo(chk.part); } } finally { p.getFileSystem(conf).delete(p, true); @@ -200,8 +134,9 @@ public void testTotalOrderBinarySearch() throws Exception { partitioner.setConf(conf); NullWritable nw = NullWritable.get(); for (Check chk : testStrings) { - assertEquals(chk.part, - partitioner.getPartition(chk.data, nw, splitStrings.length + 1), chk.data.toString()); + Assertions.assertThat(partitioner.getPartition(chk.data, nw, splitStrings.length + 1)) + .describedAs("Expected %s", chk.data) + .isEqualTo(chk.part); } } finally { p.getFileSystem(conf).delete(p, true); @@ -249,8 +184,9 @@ public void testTotalOrderCustomComparator() throws Exception { partitioner.setConf(conf); NullWritable nw = NullWritable.get(); for (Check chk : revCheck) { - assertEquals(chk.part, - partitioner.getPartition(chk.data, nw, splitStrings.length + 1), chk.data.toString()); + Assertions.assertThat(partitioner.getPartition(chk.data, nw, splitStrings.length + 1)) + .describedAs("Expected %s", chk.data) + .isEqualTo(chk.part); } } finally { p.getFileSystem(conf).delete(p, true);