hashes) {
+ String hash = hashes.get(recordCount);
if (hash == null) {
throw new UnsupportedOperationException(
- String.format("No hash for that line count: %s", lineCount)
+ String.format("No hash for that record count: %s", recordCount)
);
}
return hash;
diff --git a/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/text/TextIOIT.java b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/text/TextIOIT.java
index 3e67f6ab5803..f6e27cdb3564 100644
--- a/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/text/TextIOIT.java
+++ b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/text/TextIOIT.java
@@ -19,7 +19,7 @@
package org.apache.beam.sdk.io.text;
import static org.apache.beam.sdk.io.Compression.AUTO;
-import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.appendTimestampToPrefix;
+import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.appendTimestampSuffix;
import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.getExpectedHashForLineCount;
import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.readTestPipelineOptions;
@@ -27,6 +27,7 @@
import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.common.FileBasedIOITHelper;
+import org.apache.beam.sdk.io.common.FileBasedIOITHelper.DeleteFileFn;
import org.apache.beam.sdk.io.common.HashingFn;
import org.apache.beam.sdk.io.common.IOTestPipelineOptions;
import org.apache.beam.sdk.testing.PAssert;
@@ -75,7 +76,7 @@ public static void setup() {
IOTestPipelineOptions options = readTestPipelineOptions();
numberOfTextLines = options.getNumberOfRecords();
- filenamePrefix = appendTimestampToPrefix(options.getFilenamePrefix());
+ filenamePrefix = appendTimestampSuffix(options.getFilenamePrefix());
compressionType = Compression.valueOf(options.getCompressionType());
}
@@ -107,7 +108,7 @@ public void writeThenReadAll() {
testFilenames.apply(
"Delete test files",
- ParDo.of(new FileBasedIOITHelper.DeleteFileFn())
+ ParDo.of(new DeleteFileFn())
.withSideInputs(consolidatedHashcode.apply(View.asSingleton())));
pipeline.run().waitUntilFinish();
diff --git a/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/tfrecord/TFRecordIOIT.java b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/tfrecord/TFRecordIOIT.java
index 99d23e0f432d..1ce155912ca1 100644
--- a/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/tfrecord/TFRecordIOIT.java
+++ b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/tfrecord/TFRecordIOIT.java
@@ -19,7 +19,7 @@
package org.apache.beam.sdk.io.tfrecord;
import static org.apache.beam.sdk.io.Compression.AUTO;
-import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.appendTimestampToPrefix;
+import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.appendTimestampSuffix;
import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.getExpectedHashForLineCount;
import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.readTestPipelineOptions;
@@ -27,6 +27,7 @@
import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.TFRecordIO;
import org.apache.beam.sdk.io.common.FileBasedIOITHelper;
+import org.apache.beam.sdk.io.common.FileBasedIOITHelper.DeleteFileFn;
import org.apache.beam.sdk.io.common.HashingFn;
import org.apache.beam.sdk.io.common.IOTestPipelineOptions;
import org.apache.beam.sdk.testing.PAssert;
@@ -80,7 +81,7 @@ public static void setup() {
IOTestPipelineOptions options = readTestPipelineOptions();
numberOfTextLines = options.getNumberOfRecords();
- filenamePrefix = appendTimestampToPrefix(options.getFilenamePrefix());
+ filenamePrefix = appendTimestampSuffix(options.getFilenamePrefix());
compressionType = Compression.valueOf(options.getCompressionType());
}
@@ -121,7 +122,7 @@ public void writeThenReadAll() {
.apply(Create.of(filenamePattern))
.apply(
"Delete test files",
- ParDo.of(new FileBasedIOITHelper.DeleteFileFn())
+ ParDo.of(new DeleteFileFn())
.withSideInputs(consolidatedHashcode.apply(View.asSingleton())));
readPipeline.run().waitUntilFinish();
}
diff --git a/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/xml/XmlIOIT.java b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/xml/XmlIOIT.java
new file mode 100644
index 000000000000..7176d7f42da0
--- /dev/null
+++ b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/xml/XmlIOIT.java
@@ -0,0 +1,208 @@
+/*
+ * 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.beam.sdk.io.xml;
+
+import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.appendTimestampSuffix;
+import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.getHashForRecordCount;
+
+import com.google.common.collect.ImmutableMap;
+import java.io.Serializable;
+import java.nio.charset.Charset;
+import java.util.Map;
+import javax.xml.bind.annotation.XmlElement;
+import javax.xml.bind.annotation.XmlRootElement;
+import javax.xml.bind.annotation.XmlType;
+import org.apache.beam.sdk.io.FileIO;
+import org.apache.beam.sdk.io.GenerateSequence;
+import org.apache.beam.sdk.io.common.FileBasedIOITHelper;
+import org.apache.beam.sdk.io.common.HashingFn;
+import org.apache.beam.sdk.io.common.IOTestPipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle;
+import org.apache.beam.sdk.transforms.SimpleFunction;
+import org.apache.beam.sdk.transforms.Values;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.values.PCollection;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Integration tests for {@link org.apache.beam.sdk.io.xml.XmlIO}.
+ *
+ * Run those tests using the command below. Pass in connection information via PipelineOptions:
+ *
+ * mvn -e -Pio-it verify -pl sdks/java/io/file-based-io-tests
+ * -Dit.test=org.apache.beam.sdk.io.xml.XmlIOIT
+ * -DintegrationTestPipelineOptions='[
+ * "--numberOfRecords=100000",
+ * "--filenamePrefix=output_file_path",
+ * "--charset=UTF-8",
+ * ]'
+ *
+ *
+ */
+@RunWith(JUnit4.class)
+public class XmlIOIT {
+
+ private static final Map EXPECTED_HASHES = ImmutableMap.of(
+ 1000, "7f51adaf701441ee83459a3f705c1b86",
+ 100_000, "af7775de90d0b0c8bbc36273fbca26fe",
+ 100_000_000, "bfee52b33aa1552b9c1bfa8bcc41ae80"
+ );
+
+ private static Integer numberOfRecords;
+
+ private static String filenamePrefix;
+
+ private static Charset charset;
+
+ @Rule
+ public TestPipeline pipeline = TestPipeline.create();
+
+ @BeforeClass
+ public static void setup() {
+ PipelineOptionsFactory.register(IOTestPipelineOptions.class);
+ IOTestPipelineOptions options = TestPipeline
+ .testingPipelineOptions()
+ .as(IOTestPipelineOptions.class);
+
+ filenamePrefix = appendTimestampSuffix(options.getFilenamePrefix());
+ numberOfRecords = options.getNumberOfRecords();
+ charset = Charset.forName(options.getCharset());
+ }
+
+ @Test
+ public void writeThenReadAll() {
+ PCollection testFileNames = pipeline
+ .apply("Generate sequence", GenerateSequence.from(0).to(numberOfRecords))
+ .apply("Create xml records", MapElements.via(new LongToBird()))
+ .apply("Write xml files", FileIO.write()
+ .via(XmlIO.sink(Bird.class)
+ .withRootElement("birds")
+ .withCharset(charset))
+ .to(filenamePrefix)
+ .withPrefix("birds")
+ .withSuffix(".xml"))
+ .getPerDestinationOutputFilenames()
+ .apply("Prevent fusion", Reshuffle.viaRandomKey())
+ .apply("Get file names", Values.create());
+
+ PCollection birds = testFileNames
+ .apply("Find files", FileIO.matchAll())
+ .apply("Read matched files", FileIO.readMatches())
+ .apply("Read xml files", XmlIO.readFiles()
+ .withRecordClass(Bird.class).withRootElement("birds")
+ .withRecordElement("bird")
+ .withCharset(charset));
+
+ PCollection consolidatedHashcode = birds
+ .apply("Map xml records to strings", MapElements.via(new BirdToString()))
+ .apply("Calculate hashcode", Combine.globally(new HashingFn()));
+
+ String expectedHash = getHashForRecordCount(numberOfRecords, EXPECTED_HASHES);
+ PAssert.thatSingleton(consolidatedHashcode).isEqualTo(expectedHash);
+
+ testFileNames.apply("Delete test files", ParDo.of(new FileBasedIOITHelper.DeleteFileFn())
+ .withSideInputs(consolidatedHashcode.apply(View.asSingleton())));
+
+ pipeline.run().waitUntilFinish();
+ }
+
+ private static class LongToBird extends SimpleFunction {
+ @Override
+ public Bird apply(Long input) {
+ return new Bird("Testing", "Bird number " + input);
+ }
+ }
+
+ private static class BirdToString extends SimpleFunction {
+ @Override
+ public String apply(Bird input) {
+ return input.toString();
+ }
+ }
+
+ @SuppressWarnings("unused")
+ @XmlRootElement(name = "bird")
+ @XmlType(propOrder = { "name", "adjective" })
+ private static final class Bird implements Serializable {
+ private String name;
+ private String adjective;
+
+ @XmlElement(name = "species")
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getAdjective() {
+ return adjective;
+ }
+
+ public void setAdjective(String adjective) {
+ this.adjective = adjective;
+ }
+
+ public Bird() {}
+
+ public Bird(String adjective, String name) {
+ this.adjective = adjective;
+ this.name = name;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ Bird bird = (Bird) o;
+
+ if (!name.equals(bird.name)) {
+ return false;
+ }
+ return adjective.equals(bird.adjective);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = name.hashCode();
+ result = 31 * result + adjective.hashCode();
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("Bird: %s, %s", name, adjective);
+ }
+ }
+}
diff --git a/sdks/java/io/xml/pom.xml b/sdks/java/io/xml/pom.xml
index f4783442d75e..996063a227ce 100644
--- a/sdks/java/io/xml/pom.xml
+++ b/sdks/java/io/xml/pom.xml
@@ -109,7 +109,6 @@
hamcrest-library
test
-
diff --git a/sdks/java/io/xml/src/test/java/org/apache/beam/sdk/io/xml/XmlIOTest.java b/sdks/java/io/xml/src/test/java/org/apache/beam/sdk/io/xml/XmlIOTest.java
index f6870e12a614..cb083859b011 100644
--- a/sdks/java/io/xml/src/test/java/org/apache/beam/sdk/io/xml/XmlIOTest.java
+++ b/sdks/java/io/xml/src/test/java/org/apache/beam/sdk/io/xml/XmlIOTest.java
@@ -214,7 +214,7 @@ public void testWriteDisplayData() {
*/
@SuppressWarnings("unused")
@XmlRootElement(name = "bird")
- @XmlType(propOrder = {"name", "adjective"})
+ @XmlType(propOrder = { "name", "adjective" })
private static final class Bird implements Serializable {
private String name;
private String adjective;
@@ -266,5 +266,10 @@ public int hashCode() {
result = 31 * result + adjective.hashCode();
return result;
}
+
+ @Override
+ public String toString() {
+ return String.format("Bird: %s, %s", name, adjective);
+ }
}
}