http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/indexing/src/test/java/org/apache/rya/api/client/accumulo/FluoITBase.java ---------------------------------------------------------------------- diff --git a/extras/indexing/src/test/java/org/apache/rya/api/client/accumulo/FluoITBase.java b/extras/indexing/src/test/java/org/apache/rya/api/client/accumulo/FluoITBase.java index 695704b..7277fe0 100644 --- a/extras/indexing/src/test/java/org/apache/rya/api/client/accumulo/FluoITBase.java +++ b/extras/indexing/src/test/java/org/apache/rya/api/client/accumulo/FluoITBase.java @@ -33,11 +33,28 @@ import org.apache.accumulo.core.client.Instance; import org.apache.accumulo.core.client.ZooKeeperInstance; import org.apache.accumulo.core.client.security.tokens.PasswordToken; import org.apache.accumulo.minicluster.MiniAccumuloCluster; +import org.apache.fluo.api.client.FluoAdmin; +import org.apache.fluo.api.client.FluoAdmin.AlreadyInitializedException; +import org.apache.fluo.api.client.FluoAdmin.TableExistsException; +import org.apache.fluo.api.client.FluoClient; +import org.apache.fluo.api.client.FluoFactory; +import org.apache.fluo.api.config.FluoConfiguration; +import org.apache.fluo.api.config.ObserverSpecification; +import org.apache.fluo.api.mini.MiniFluo; import org.apache.log4j.Level; import org.apache.log4j.Logger; +import org.apache.rya.accumulo.AccumuloRdfConfiguration; import org.apache.rya.accumulo.MiniAccumuloClusterInstance; import org.apache.rya.accumulo.MiniAccumuloSingleton; import org.apache.rya.accumulo.RyaTestInstanceRule; +import org.apache.rya.api.client.Install; +import org.apache.rya.api.client.Install.DuplicateInstanceNameException; +import org.apache.rya.api.client.Install.InstallConfiguration; +import org.apache.rya.api.client.RyaClientException; +import org.apache.rya.api.instance.RyaDetailsRepository.RyaDetailsRepositoryException; +import org.apache.rya.api.persist.RyaDAOException; +import org.apache.rya.indexing.accumulo.ConfigUtils; +import org.apache.rya.indexing.external.PrecomputedJoinIndexerConfig; import org.apache.rya.indexing.pcj.fluo.app.export.rya.RyaExportParameters; import org.apache.rya.indexing.pcj.fluo.app.observers.ConstructQueryResultObserver; import org.apache.rya.indexing.pcj.fluo.app.observers.FilterObserver; @@ -46,6 +63,9 @@ import org.apache.rya.indexing.pcj.fluo.app.observers.ProjectionObserver; import org.apache.rya.indexing.pcj.fluo.app.observers.QueryResultObserver; import org.apache.rya.indexing.pcj.fluo.app.observers.StatementPatternObserver; import org.apache.rya.indexing.pcj.fluo.app.observers.TripleObserver; +import org.apache.rya.rdftriplestore.RyaSailRepository; +import org.apache.rya.rdftriplestore.inference.InferenceEngineException; +import org.apache.rya.sail.config.RyaSailFactory; import org.apache.zookeeper.ClientCnxn; import org.junit.After; import org.junit.Before; @@ -56,27 +76,6 @@ import org.openrdf.repository.RepositoryException; import org.openrdf.sail.Sail; import org.openrdf.sail.SailException; -import org.apache.fluo.api.client.FluoAdmin; -import org.apache.fluo.api.client.FluoAdmin.AlreadyInitializedException; -import org.apache.fluo.api.client.FluoAdmin.TableExistsException; -import org.apache.fluo.api.client.FluoClient; -import org.apache.fluo.api.client.FluoFactory; -import org.apache.fluo.api.config.FluoConfiguration; -import org.apache.fluo.api.config.ObserverSpecification; -import org.apache.fluo.api.mini.MiniFluo; -import org.apache.rya.accumulo.AccumuloRdfConfiguration; -import org.apache.rya.api.client.RyaClientException; -import org.apache.rya.api.client.Install; -import org.apache.rya.api.client.Install.DuplicateInstanceNameException; -import org.apache.rya.api.client.Install.InstallConfiguration; -import org.apache.rya.api.instance.RyaDetailsRepository.RyaDetailsRepositoryException; -import org.apache.rya.api.persist.RyaDAOException; -import org.apache.rya.indexing.accumulo.ConfigUtils; -import org.apache.rya.indexing.external.PrecomputedJoinIndexerConfig; -import org.apache.rya.rdftriplestore.RyaSailRepository; -import org.apache.rya.rdftriplestore.inference.InferenceEngineException; -import org.apache.rya.sail.config.RyaSailFactory; - /** * Integration tests that ensure the Fluo application processes PCJs results * correctly. @@ -249,6 +248,8 @@ public abstract class FluoITBase { return FluoFactory.newMiniFluo(config); } + static final AccumuloRdfConfiguration conf = new AccumuloRdfConfiguration(); + /** * Sets up a Rya instance. */ @@ -260,7 +261,6 @@ public abstract class FluoITBase { checkNotNull(zookeepers); // Setup Rya configuration values. - final AccumuloRdfConfiguration conf = new AccumuloRdfConfiguration(); conf.setTablePrefix(getRyaInstanceName()); conf.setDisplayQueryPlan(true); conf.setBoolean(ConfigUtils.USE_MOCK_INSTANCE, false);
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/indexing/src/test/java/org/apache/rya/indexing/accumulo/entity/StarQueryTest.java ---------------------------------------------------------------------- diff --git a/extras/indexing/src/test/java/org/apache/rya/indexing/accumulo/entity/StarQueryTest.java b/extras/indexing/src/test/java/org/apache/rya/indexing/accumulo/entity/StarQueryTest.java index babc83e..993c758 100644 --- a/extras/indexing/src/test/java/org/apache/rya/indexing/accumulo/entity/StarQueryTest.java +++ b/extras/indexing/src/test/java/org/apache/rya/indexing/accumulo/entity/StarQueryTest.java @@ -1,5 +1,3 @@ -package org.apache.rya.indexing.accumulo.entity; - /* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file @@ -8,9 +6,9 @@ package org.apache.rya.indexing.accumulo.entity; * 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 @@ -18,22 +16,18 @@ package org.apache.rya.indexing.accumulo.entity; * specific language governing permissions and limitations * under the License. */ - - -import static org.junit.Assert.*; +package org.apache.rya.indexing.accumulo.entity; import java.util.List; import java.util.Map; import java.util.Set; +import org.apache.hadoop.io.Text; import org.apache.rya.accumulo.documentIndex.TextColumn; import org.apache.rya.api.domain.RyaType; import org.apache.rya.api.resolver.RdfToRyaConversions; import org.apache.rya.api.resolver.RyaContext; import org.apache.rya.api.resolver.RyaTypeResolverException; -import org.apache.rya.indexing.accumulo.entity.StarQuery; - -import org.apache.hadoop.io.Text; import org.junit.Assert; import org.junit.Test; import org.openrdf.model.Value; @@ -57,7 +51,7 @@ public class StarQueryTest { @Test - public void testBasicFunctionality() { + public void testBasicFunctionality() throws MalformedQueryException { String q1 = "" // + "SELECT ?X ?Y1 ?Y2 " // @@ -73,11 +67,7 @@ public class StarQueryTest { SPARQLParser parser = new SPARQLParser(); ParsedQuery pq1 = null; - try { - pq1 = parser.parseQuery(q1, null); - } catch (MalformedQueryException e) { - e.printStackTrace(); - } + pq1 = parser.parseQuery(q1, null); TupleExpr te1 = pq1.getTupleExpr(); @@ -158,7 +148,7 @@ public class StarQueryTest { @Test - public void testGetContrainedQuery() { + public void testGetContrainedQuery() throws RyaTypeResolverException, MalformedQueryException { String q1 = "" // + "SELECT ?X ?Y1 ?Y2 " // @@ -174,12 +164,7 @@ public class StarQueryTest { SPARQLParser parser = new SPARQLParser(); ParsedQuery pq1 = null; - try { - pq1 = parser.parseQuery(q1, null); - } catch (MalformedQueryException e) { - // TODO Auto-generated catch block - e.printStackTrace(); - } + pq1 = parser.parseQuery(q1, null); TupleExpr te1 = pq1.getTupleExpr(); @@ -221,11 +206,7 @@ public class StarQueryTest { Assert.assertEquals(tc3[i].getColumnFamily(), new Text("uri:cf3")); RyaType objType = RdfToRyaConversions.convertValue(v2); byte[][] b1 = null; - try { b1 = RyaContext.getInstance().serializeType(objType); - } catch (RyaTypeResolverException e) { - e.printStackTrace(); - } byte[] b2 = Bytes.concat("object".getBytes(), "\u0000".getBytes(), b1[0], b1[1]); Assert.assertEquals(tc3[i].getColumnQualifier(), new Text(b2)); @@ -241,7 +222,7 @@ public class StarQueryTest { @Test - public void testConstantPriority() { + public void testConstantPriority() throws MalformedQueryException { String q1 = "" // + "SELECT ?X " // @@ -257,11 +238,7 @@ public class StarQueryTest { SPARQLParser parser = new SPARQLParser(); ParsedQuery pq1 = null; - try { - pq1 = parser.parseQuery(q1, null); - } catch (MalformedQueryException e) { - e.printStackTrace(); - } + pq1 = parser.parseQuery(q1, null); TupleExpr te1 = pq1.getTupleExpr(); http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/indexingExample/src/main/java/InferenceExamples.java ---------------------------------------------------------------------- diff --git a/extras/indexingExample/src/main/java/InferenceExamples.java b/extras/indexingExample/src/main/java/InferenceExamples.java index 5b5a937..83c3bc0 100644 --- a/extras/indexingExample/src/main/java/InferenceExamples.java +++ b/extras/indexingExample/src/main/java/InferenceExamples.java @@ -73,7 +73,7 @@ public class InferenceExamples { private static final String MONGO_INSTANCE_URL = "localhost"; private static final String MONGO_INSTANCE_PORT = "27017"; private static final String MongoUserName="usern"; - private static final String MongoUserPassword="passwd"; + private static final String MongoUserPswd="passwd"; public static void setupLogging() { final Logger rootLogger = LogManager.getRootLogger(); @@ -156,7 +156,7 @@ public class InferenceExamples { } else { // User name and password must be filled in: builder = builder.setMongoUser(MongoUserName) - .setMongoPassword(MongoUserPassword) + .setMongoPassword(MongoUserPswd) .setMongoHost(MONGO_INSTANCE_URL) .setMongoPort(MONGO_INSTANCE_PORT); } http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/periodic.notification/twill.yarn/src/main/java/org/apache/rya/periodic/notification/twill/yarn/PeriodicNotificationTwillRunner.java ---------------------------------------------------------------------- diff --git a/extras/periodic.notification/twill.yarn/src/main/java/org/apache/rya/periodic/notification/twill/yarn/PeriodicNotificationTwillRunner.java b/extras/periodic.notification/twill.yarn/src/main/java/org/apache/rya/periodic/notification/twill/yarn/PeriodicNotificationTwillRunner.java index 552324e..a91d1e3 100644 --- a/extras/periodic.notification/twill.yarn/src/main/java/org/apache/rya/periodic/notification/twill/yarn/PeriodicNotificationTwillRunner.java +++ b/extras/periodic.notification/twill.yarn/src/main/java/org/apache/rya/periodic/notification/twill/yarn/PeriodicNotificationTwillRunner.java @@ -284,12 +284,16 @@ public class PeriodicNotificationTwillRunner implements AutoCloseable { } // pick the correct zookeepers - String zookeepers; + String zookeepers = null; if(options.zookeepers != null && !options.zookeepers.isEmpty()) { zookeepers = options.zookeepers; - } else { + } else if (conf != null) { zookeepers = conf.getAccumuloZookeepers(); } + if (zookeepers == null) { + LOG.warn("Zookeeper connection info can not be determined from main options nor configuration file."); + System.exit(1); + } try (final PeriodicNotificationTwillRunner app = new PeriodicNotificationTwillRunner(zookeepers, options.configFile)) { if(START.equals(parsedCommand)) { http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.benchmark/src/main/java/org/apache/rya/benchmark/query/QueriesBenchmarkConfReader.java ---------------------------------------------------------------------- diff --git a/extras/rya.benchmark/src/main/java/org/apache/rya/benchmark/query/QueriesBenchmarkConfReader.java b/extras/rya.benchmark/src/main/java/org/apache/rya/benchmark/query/QueriesBenchmarkConfReader.java index ce91ba1..12440da 100644 --- a/extras/rya.benchmark/src/main/java/org/apache/rya/benchmark/query/QueriesBenchmarkConfReader.java +++ b/extras/rya.benchmark/src/main/java/org/apache/rya/benchmark/query/QueriesBenchmarkConfReader.java @@ -20,23 +20,29 @@ package org.apache.rya.benchmark.query; import static java.util.Objects.requireNonNull; +import java.io.IOException; import java.io.InputStream; -import edu.umd.cs.findbugs.annotations.DefaultAnnotation; -import edu.umd.cs.findbugs.annotations.NonNull; import javax.xml.XMLConstants; import javax.xml.bind.JAXBContext; import javax.xml.bind.JAXBException; import javax.xml.bind.Unmarshaller; +import javax.xml.parsers.DocumentBuilder; +import javax.xml.parsers.DocumentBuilderFactory; +import javax.xml.parsers.ParserConfigurationException; import javax.xml.transform.stream.StreamSource; import javax.xml.validation.Schema; import javax.xml.validation.SchemaFactory; +import org.apache.rya.api.utils.XmlFactoryConfiguration; import org.xml.sax.SAXException; import com.google.common.base.Supplier; import com.google.common.base.Suppliers; +import edu.umd.cs.findbugs.annotations.DefaultAnnotation; +import edu.umd.cs.findbugs.annotations.NonNull; + /** * Unmarshalls instances of {@link QueriesBenchmarkConf}. */ @@ -57,6 +63,14 @@ public final class QueriesBenchmarkConfReader { return schemaFactory.newSchema( new StreamSource( schemaStream ) ); } catch (final SAXException e) { throw new RuntimeException("Could not load the '" + SCHEMA_LOCATION + "' schema file. Make sure it is on the classpath.", e); + } finally { + if (schemaStream != null) { + try { + schemaStream.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } } } }); @@ -68,8 +82,11 @@ public final class QueriesBenchmarkConfReader { * @param xmlStream - The input stream holding the XML. (not null) * @return The {@link BenchmarkQueries} instance that was read from the stream. * @throws JAXBException There was a problem with the formatting of the XML. + * @throws ParserConfigurationException There was a problem creating the DocumentBuilder. + * @throws IOException There was a problem reading the xmlStream. + * @throws SAXException There was a problem parsing the xmlStream. */ - public QueriesBenchmarkConf load(final InputStream xmlStream) throws JAXBException { + public QueriesBenchmarkConf load(final InputStream xmlStream) throws JAXBException, ParserConfigurationException, SAXException, IOException { requireNonNull(xmlStream); // Load the schema that describes the stream. @@ -79,6 +96,9 @@ public final class QueriesBenchmarkConfReader { final JAXBContext context = JAXBContext.newInstance( QueriesBenchmarkConf.class ); final Unmarshaller unmarshaller = context.createUnmarshaller(); unmarshaller.setSchema(schema); - return (QueriesBenchmarkConf) unmarshaller.unmarshal(xmlStream); + final DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance(); + XmlFactoryConfiguration.harden(dbf); + final DocumentBuilder db = dbf.newDocumentBuilder(); + return (QueriesBenchmarkConf) unmarshaller.unmarshal(db.parse(xmlStream)); } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.benchmark/src/main/xsd/queries-benchmark-conf.xsd ---------------------------------------------------------------------- diff --git a/extras/rya.benchmark/src/main/xsd/queries-benchmark-conf.xsd b/extras/rya.benchmark/src/main/xsd/queries-benchmark-conf.xsd index 826083e..a827432 100644 --- a/extras/rya.benchmark/src/main/xsd/queries-benchmark-conf.xsd +++ b/extras/rya.benchmark/src/main/xsd/queries-benchmark-conf.xsd @@ -58,14 +58,14 @@ <xsd:element name="NumReadsRuns"> <xsd:complexType> <xsd:sequence> - <xsd:element name="NumReads" type="xsd:string" maxOccurs="unbounded"/> + <xsd:element name="NumReads" type="xsd:string" maxOccurs="32766"/> </xsd:sequence> </xsd:complexType> </xsd:element> <xsd:element name="Queries" minOccurs="0"> <xsd:complexType> <xsd:sequence> - <xsd:element name="SPARQL" type="xsd:string" maxOccurs="unbounded"/> + <xsd:element name="SPARQL" type="xsd:string" maxOccurs="32766"/> </xsd:sequence> </xsd:complexType> </xsd:element> http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.benchmark/src/test/java/org/apache/rya/benchmark/query/QueriesBenchmarkConfReaderIT.java ---------------------------------------------------------------------- diff --git a/extras/rya.benchmark/src/test/java/org/apache/rya/benchmark/query/QueriesBenchmarkConfReaderIT.java b/extras/rya.benchmark/src/test/java/org/apache/rya/benchmark/query/QueriesBenchmarkConfReaderIT.java index f229dc4..4b8ec80 100644 --- a/extras/rya.benchmark/src/test/java/org/apache/rya/benchmark/query/QueriesBenchmarkConfReaderIT.java +++ b/extras/rya.benchmark/src/test/java/org/apache/rya/benchmark/query/QueriesBenchmarkConfReaderIT.java @@ -22,10 +22,12 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import java.io.ByteArrayInputStream; +import java.io.IOException; import java.io.InputStream; import java.util.List; import javax.xml.bind.JAXBException; +import javax.xml.parsers.ParserConfigurationException; import org.apache.rya.benchmark.query.Parameters.NumReadsRuns; import org.apache.rya.benchmark.query.Parameters.Queries; @@ -43,7 +45,7 @@ import com.google.common.collect.Lists; public class QueriesBenchmarkConfReaderIT { @Test - public void load() throws JAXBException, SAXException { + public void load() throws JAXBException, SAXException, ParserConfigurationException, IOException { // Unmarshal some XML. final String xml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n" + http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/AccumuloRyaStatementStore.java ---------------------------------------------------------------------- diff --git a/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/AccumuloRyaStatementStore.java b/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/AccumuloRyaStatementStore.java index 2376c20..f5992df 100644 --- a/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/AccumuloRyaStatementStore.java +++ b/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/AccumuloRyaStatementStore.java @@ -184,9 +184,12 @@ public class AccumuloRyaStatementStore implements RyaStatementStore { MergeParentMetadata metadata = null; try { metadata = metadataRepo.get(); - } finally { - return Optional.ofNullable(metadata); + } catch (final Exception e) { + // Catching any exception to ensure we always return Optional.ofNullable(metadata). + // Logging at the debug level if exceptional behavior needs to be investigated while deployed. + log.debug("Parent metadata missing or exceptional behavior occurred.", e); } + return Optional.ofNullable(metadata); } @Override http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/parent/AccumuloParentMetadataRepository.java ---------------------------------------------------------------------- diff --git a/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/parent/AccumuloParentMetadataRepository.java b/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/parent/AccumuloParentMetadataRepository.java index 3241fbb..5f50a43 100644 --- a/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/parent/AccumuloParentMetadataRepository.java +++ b/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/parent/AccumuloParentMetadataRepository.java @@ -135,7 +135,7 @@ public class AccumuloParentMetadataRepository implements ParentMetadataRepositor // No metadata has been stored in the table yet. if (!entries.hasNext()) { - log.error("Could not find any MergeParentMetadata metadata in the table named: " + mergeParentMetadataTableName); + throw new ParentMetadataDoesNotExistException("Could not find any MergeParentMetadata metadata in the table named: " + mergeParentMetadataTableName); } // Fetch the metadata from the entries. http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/util/AccumuloInstanceDriver.java ---------------------------------------------------------------------- diff --git a/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/util/AccumuloInstanceDriver.java b/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/util/AccumuloInstanceDriver.java index f8d9484..e077751 100644 --- a/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/util/AccumuloInstanceDriver.java +++ b/extras/rya.export/export.accumulo/src/main/java/org/apache/rya/export/accumulo/util/AccumuloInstanceDriver.java @@ -47,6 +47,7 @@ import org.apache.rya.accumulo.AccumuloRdfConfiguration; import org.apache.rya.accumulo.AccumuloRyaDAO; import org.apache.rya.accumulo.mr.MRUtils; import org.apache.rya.api.RdfCloudTripleStoreConstants; +import org.apache.rya.api.path.PathUtils; import org.apache.rya.api.persist.RyaDAOException; import org.apache.rya.export.InstanceType; import org.apache.rya.export.accumulo.conf.AccumuloExportConstants; @@ -73,7 +74,7 @@ public class AccumuloInstanceDriver { private final boolean isParent; private final String user; - private final String password; + private final String userpwd; private final String instanceName; private final String tablePrefix; private final String auth; @@ -117,29 +118,41 @@ public class AccumuloInstanceDriver { /** * Creates a new instance of {@link AccumuloInstanceDriver}. - * @param driverName the name used to identify this driver in the logs. (not {@code null}) - * @param instanceType the {@link InstanceType} of this driver. - * @param shouldCreateIndices {@code true} to create all the indices associated with a Rya deployment. - * {@code false} otherwise. - * @param isReadOnly {@code true} if all the tables in the instance should have their - * table permissions set to read only. {@code false} if the table permission are set to write. - * @param isParent {@code true} if the instance is the parent/main instance. {@code false} if it's the - * child. - * @param user the user name tied to this instance. - * @param password the password for the user. - * @param instanceName the name of the instance. - * @param tablePrefix the table prefix. - * @param auth the comma-separated authorization list. - * @param zooKeepers the comma-separated list of zoo keeper host names. - */ - public AccumuloInstanceDriver(final String driverName, final InstanceType instanceType, final boolean shouldCreateIndices, final boolean isReadOnly, final boolean isParent, final String user, final String password, final String instanceName, final String tablePrefix, final String auth, final String zooKeepers) { + * + * @param driverName + * the name used to identify this driver in the logs. (not {@code null}) + * @param instanceType + * the {@link InstanceType} of this driver. + * @param shouldCreateIndices + * {@code true} to create all the indices associated with a Rya deployment. + * {@code false} otherwise. + * @param isReadOnly + * {@code true} if all the tables in the instance should have their + * table permissions set to read only. {@code false} if the table permission are set to write. + * @param isParent + * {@code true} if the instance is the parent/main instance. {@code false} if it's the + * child. + * @param user + * the user name tied to this instance. + * @param userpwd + * the password for the user. + * @param instanceName + * the name of the instance. + * @param tablePrefix + * the table prefix. + * @param auth + * the comma-separated authorization list. + * @param zooKeepers + * the comma-separated list of zoo keeper host names. + */ + public AccumuloInstanceDriver(final String driverName, final InstanceType instanceType, final boolean shouldCreateIndices, final boolean isReadOnly, final boolean isParent, final String user, final String userpwd, final String instanceName, final String tablePrefix, final String auth, final String zooKeepers) { this.driverName = Preconditions.checkNotNull(driverName); this.instanceType = instanceType; isMock = instanceType == InstanceType.MOCK; this.shouldCreateIndices = shouldCreateIndices; this.isReadOnly = isReadOnly; this.user = user; - this.password = password; + this.userpwd = userpwd; this.instanceName = instanceName; this.tablePrefix = tablePrefix; this.auth = auth; @@ -177,7 +190,7 @@ public class AccumuloInstanceDriver { throw new IllegalArgumentException("Must specify ZooKeeper hosts for distributed mode"); } instance = new ZooKeeperInstance(instanceName, zooKeepers); - connector = instance.getConnector(user, new PasswordToken(password)); + connector = instance.getConnector(user, new PasswordToken(userpwd)); log.info("Created connector to " + driverName + " distribution instance"); break; case MINI: @@ -185,7 +198,7 @@ public class AccumuloInstanceDriver { // Create and Run MiniAccumulo Cluster tempDir = Files.createTempDir(); tempDir.deleteOnExit(); - miniAccumuloCluster = new MiniAccumuloCluster(tempDir, password); + miniAccumuloCluster = new MiniAccumuloCluster(tempDir, userpwd); copyHadoopHomeToTemp(); miniAccumuloCluster.getConfig().setInstanceName(instanceName); log.info(driverName + " MiniAccumulo instance starting up..."); @@ -195,14 +208,14 @@ public class AccumuloInstanceDriver { log.info("Creating connector to " + driverName + " MiniAccumulo instance..."); zooKeeperInstance = new ZooKeeperInstance(miniAccumuloCluster.getClientConfig()); instance = zooKeeperInstance; - connector = zooKeeperInstance.getConnector(user, new PasswordToken(password)); + connector = zooKeeperInstance.getConnector(user, new PasswordToken(userpwd)); log.info("Created connector to " + driverName + " MiniAccumulo instance"); break; case MOCK: log.info("Setting up " + driverName + " mock instance..."); mockInstance = new MockInstance(instanceName); instance = mockInstance; - connector = mockInstance.getConnector(user, new PasswordToken(password)); + connector = mockInstance.getConnector(user, new PasswordToken(userpwd)); log.info("Created connector to " + driverName + " mock instance"); break; default: @@ -219,7 +232,7 @@ public class AccumuloInstanceDriver { */ private void copyHadoopHomeToTemp() throws IOException { if (IS_COPY_HADOOP_HOME_ENABLED && SystemUtils.IS_OS_WINDOWS) { - final String hadoopHomeEnv = System.getenv("HADOOP_HOME"); + final String hadoopHomeEnv = PathUtils.clean(System.getenv("HADOOP_HOME")); if (hadoopHomeEnv != null) { final File hadoopHomeDir = new File(hadoopHomeEnv); if (hadoopHomeDir.exists()) { @@ -278,7 +291,7 @@ public class AccumuloInstanceDriver { log.info("Creating " + driverName + " user and authorizations"); secOps = connector.securityOperations(); if (!user.equals(ROOT_USER_NAME)) { - secOps.createLocalUser(user, new PasswordToken(password)); + secOps.createLocalUser(user, new PasswordToken(userpwd)); } addAuths(auth); final TablePermission tablePermission = isReadOnly ? TablePermission.READ : TablePermission.WRITE; @@ -325,7 +338,7 @@ public class AccumuloInstanceDriver { } configMap.put(MRUtils.AC_INSTANCE_PROP, instanceName); configMap.put(MRUtils.AC_USERNAME_PROP, user); - configMap.put(MRUtils.AC_PWD_PROP, password); + configMap.put(MRUtils.AC_PWD_PROP, userpwd); configMap.put(MRUtils.TABLE_PREFIX_PROPERTY, tablePrefix); configMap.put(MRUtils.AC_AUTH_PROP, auth); configMap.put(MRUtils.AC_ZK_PROP, zooKeepers != null ? zooKeepers : "localhost"); @@ -337,7 +350,9 @@ public class AccumuloInstanceDriver { final String key = entry.getKey(); final String value = entry.getValue(); final String argument = "-D" + key + "=" + value; - log.info(argument); + if (!key.equals(MRUtils.AC_PWD_PROP)) { + log.info(argument); + } config.set(key, value); } @@ -481,10 +496,10 @@ public class AccumuloInstanceDriver { } /** - * @return the password for the user. + * @return the userpwd for the user. */ public String getPassword() { - return password; + return userpwd; } /** http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.export/export.client/conf/config.xml ---------------------------------------------------------------------- diff --git a/extras/rya.export/export.client/conf/config.xml b/extras/rya.export/export.client/conf/config.xml index f2a7fdd..879240c 100644 --- a/extras/rya.export/export.client/conf/config.xml +++ b/extras/rya.export/export.client/conf/config.xml @@ -32,5 +32,4 @@ under the License. --> <childPort>27017</childPort> <mergePolicy>timestamp</mergePolicy> <useNtpServer>false</useNtpServer> - <toolStartTime>Sep 026 2016 15:21:30</toolStartTime> </MergeToolConfiguration> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/MergeDriverClient.java ---------------------------------------------------------------------- diff --git a/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/MergeDriverClient.java b/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/MergeDriverClient.java index 992cc09..232a0ed 100644 --- a/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/MergeDriverClient.java +++ b/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/MergeDriverClient.java @@ -32,6 +32,7 @@ import org.apache.commons.lang.StringUtils; import org.apache.log4j.BasicConfigurator; import org.apache.log4j.Logger; import org.apache.log4j.xml.DOMConfigurator; +import org.apache.rya.api.path.PathUtils; import org.apache.rya.export.accumulo.AccumuloRyaStatementStore; import org.apache.rya.export.api.MergerException; import org.apache.rya.export.api.conf.MergeConfiguration; @@ -66,7 +67,7 @@ public class MergeDriverClient { final String log4jConfiguration = System.getProperties().getProperty("log4j.configuration"); if (StringUtils.isNotBlank(log4jConfiguration)) { - final String parsedConfiguration = StringUtils.removeStart(log4jConfiguration, "file:"); + final String parsedConfiguration = PathUtils.clean(StringUtils.removeStart(log4jConfiguration, "file:")); final File configFile = new File(parsedConfiguration); if (configFile.exists()) { DOMConfigurator.configure(parsedConfiguration); http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/conf/MergeConfigurationCLI.java ---------------------------------------------------------------------- diff --git a/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/conf/MergeConfigurationCLI.java b/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/conf/MergeConfigurationCLI.java index b8a64d9..8f39bd3 100644 --- a/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/conf/MergeConfigurationCLI.java +++ b/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/conf/MergeConfigurationCLI.java @@ -22,6 +22,7 @@ import static com.google.common.base.Preconditions.checkNotNull; import static org.apache.rya.export.MergePolicy.TIMESTAMP; import java.io.File; +import java.io.IOException; import java.text.DateFormat; import java.text.SimpleDateFormat; import java.util.Date; @@ -29,6 +30,9 @@ import java.util.Date; import javax.xml.bind.JAXBContext; import javax.xml.bind.JAXBException; import javax.xml.bind.Unmarshaller; +import javax.xml.parsers.DocumentBuilder; +import javax.xml.parsers.DocumentBuilderFactory; +import javax.xml.parsers.ParserConfigurationException; import org.apache.commons.cli.BasicParser; import org.apache.commons.cli.CommandLine; @@ -36,6 +40,7 @@ import org.apache.commons.cli.CommandLineParser; import org.apache.commons.cli.Option; import org.apache.commons.cli.Options; import org.apache.commons.cli.ParseException; +import org.apache.rya.api.utils.XmlFactoryConfiguration; import org.apache.rya.export.AccumuloMergeToolConfiguration; import org.apache.rya.export.DBType; import org.apache.rya.export.InstanceType; @@ -47,6 +52,7 @@ import org.apache.rya.export.api.conf.ConfigurationAdapter; import org.apache.rya.export.api.conf.MergeConfiguration; import org.apache.rya.export.api.conf.MergeConfigurationException; import org.apache.rya.export.api.conf.policy.TimestampPolicyMergeConfiguration; +import org.xml.sax.SAXException; import com.google.common.annotations.VisibleForTesting; @@ -140,8 +146,11 @@ public class MergeConfigurationCLI { try { final JAXBContext context = JAXBContext.newInstance(DBType.class, MergeToolConfiguration.class, AccumuloMergeToolConfiguration.class, TimestampMergePolicyConfiguration.class, MergePolicy.class, InstanceType.class); final Unmarshaller unmarshaller = context.createUnmarshaller(); - return (MergeToolConfiguration) unmarshaller.unmarshal(configFile); - } catch (final JAXBException | IllegalArgumentException JAXBe) { + final DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance(); + XmlFactoryConfiguration.harden(dbf); + final DocumentBuilder db = dbf.newDocumentBuilder(); + return unmarshaller.unmarshal(db.parse(configFile), MergeToolConfiguration.class).getValue(); + } catch (final JAXBException | IllegalArgumentException | ParserConfigurationException | SAXException | IOException JAXBe) { throw new MergeConfigurationException("Failed to create a config based on the provided configuration.", JAXBe); } } http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/conf/TimeUtils.java ---------------------------------------------------------------------- diff --git a/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/conf/TimeUtils.java b/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/conf/TimeUtils.java index c7e2174..b1731de 100644 --- a/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/conf/TimeUtils.java +++ b/extras/rya.export/export.client/src/main/java/org/apache/rya/export/client/conf/TimeUtils.java @@ -128,7 +128,9 @@ public final class TimeUtils { } } finally { // Close the connection - conn.disconnect(); + if (conn != null) { + conn.disconnect(); + } } return remoteDate; http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.export/export.client/src/test/java/org/apache/rya/export/client/conf/MergeConfigurationCLITest.java ---------------------------------------------------------------------- diff --git a/extras/rya.export/export.client/src/test/java/org/apache/rya/export/client/conf/MergeConfigurationCLITest.java b/extras/rya.export/export.client/src/test/java/org/apache/rya/export/client/conf/MergeConfigurationCLITest.java new file mode 100644 index 0000000..75ad021 --- /dev/null +++ b/extras/rya.export/export.client/src/test/java/org/apache/rya/export/client/conf/MergeConfigurationCLITest.java @@ -0,0 +1,56 @@ +/* + * 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.rya.export.client.conf; + +import static org.junit.Assert.assertEquals; + +import java.io.File; + +import javax.xml.bind.JAXBException; + +import org.apache.rya.export.DBType; +import org.apache.rya.export.MergePolicy; +import org.apache.rya.export.MergeToolConfiguration; +import org.apache.rya.export.api.conf.MergeConfigurationException; +import org.junit.Test; + +public class MergeConfigurationCLITest { + @Test + public void testCreate1ConfigurationFromFile() throws MergeConfigurationException, JAXBException { + + final MergeToolConfiguration conf = MergeConfigurationCLI.createConfigurationFromFile(new File("conf/config.xml")); + assertEquals("10.10.10.100", conf.getParentHostname()); + assertEquals("accumuloUsername", conf.getParentUsername()); + assertEquals("accumuloPassword", conf.getParentPassword()); + assertEquals("accumuloInstance", conf.getParentRyaInstanceName()); + assertEquals("rya_demo_export_", conf.getParentTablePrefix()); + assertEquals("http://10.10.10.100:8080";, conf.getParentTomcatUrl()); + assertEquals(DBType.ACCUMULO, conf.getParentDBType()); + assertEquals(1111, conf.getParentPort()); + assertEquals("10.10.10.101", conf.getChildHostname()); + assertEquals("rya_demo_child", conf.getChildRyaInstanceName()); + assertEquals("rya_demo_export_", conf.getChildTablePrefix()); + assertEquals("http://10.10.10.101:8080";, conf.getChildTomcatUrl()); + assertEquals(DBType.MONGO, conf.getChildDBType()); + assertEquals(27017, conf.getChildPort()); + assertEquals(MergePolicy.TIMESTAMP, conf.getMergePolicy()); + assertEquals(Boolean.FALSE, conf.isUseNtpServer()); + assertEquals(null, conf.getNtpServerHost()); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.export/export.mongo/src/main/java/org/apache/rya/export/mongo/MongoRyaStatementStore.java ---------------------------------------------------------------------- diff --git a/extras/rya.export/export.mongo/src/main/java/org/apache/rya/export/mongo/MongoRyaStatementStore.java b/extras/rya.export/export.mongo/src/main/java/org/apache/rya/export/mongo/MongoRyaStatementStore.java index a25f6b2..d77c0d6 100644 --- a/extras/rya.export/export.mongo/src/main/java/org/apache/rya/export/mongo/MongoRyaStatementStore.java +++ b/extras/rya.export/export.mongo/src/main/java/org/apache/rya/export/mongo/MongoRyaStatementStore.java @@ -38,6 +38,8 @@ import org.apache.rya.export.api.store.UpdateStatementException; import org.apache.rya.export.mongo.parent.MongoParentMetadataRepository; import org.apache.rya.mongodb.MongoDBRyaDAO; import org.apache.rya.mongodb.dao.SimpleMongoDBStorageStrategy; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import com.mongodb.BasicDBObject; import com.mongodb.Cursor; @@ -50,6 +52,8 @@ import com.mongodb.MongoClient; * importing rya statements from MongoDB. */ public class MongoRyaStatementStore implements RyaStatementStore { + private static final Logger logger = LoggerFactory.getLogger(MongoRyaStatementStore.class); + public static final String TRIPLES_COLLECTION = "rya__triples"; public static final String METADATA_COLLECTION = "parent_metadata"; protected final SimpleMongoDBStorageStrategy adapter; @@ -130,9 +134,12 @@ public class MongoRyaStatementStore implements RyaStatementStore { MergeParentMetadata metadata = null; try { metadata = parentMetadataRepo.get(); - } finally { - return Optional.ofNullable(metadata); + } catch (final Exception e) { + // Catching any exception to ensure we always return Optional.ofNullable(metadata). + // Logging at the debug level if exceptional behavior needs to be investigated while deployed. + logger.debug("Parent metadata missing or exceptional behavior occurred.", e); } + return Optional.ofNullable(metadata); } @Override http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.geoindexing/geo.geomesa/src/main/java/org/apache/rya/indexing/accumulo/geo/GeoMesaGeoIndexer.java ---------------------------------------------------------------------- diff --git a/extras/rya.geoindexing/geo.geomesa/src/main/java/org/apache/rya/indexing/accumulo/geo/GeoMesaGeoIndexer.java b/extras/rya.geoindexing/geo.geomesa/src/main/java/org/apache/rya/indexing/accumulo/geo/GeoMesaGeoIndexer.java index 02ef5ba..b029b02 100644 --- a/extras/rya.geoindexing/geo.geomesa/src/main/java/org/apache/rya/indexing/accumulo/geo/GeoMesaGeoIndexer.java +++ b/extras/rya.geoindexing/geo.geomesa/src/main/java/org/apache/rya/indexing/accumulo/geo/GeoMesaGeoIndexer.java @@ -41,6 +41,7 @@ import org.apache.log4j.Logger; import org.apache.rya.accumulo.experimental.AbstractAccumuloIndexer; import org.apache.rya.api.RdfCloudTripleStoreConfiguration; import org.apache.rya.api.domain.RyaStatement; +import org.apache.rya.api.log.LogUtils; import org.apache.rya.api.resolver.RyaToRdfConversions; import org.apache.rya.indexing.GeoIndexer; import org.apache.rya.indexing.Md5Hash; @@ -62,6 +63,7 @@ import org.geotools.feature.SchemaException; import org.geotools.feature.simple.SimpleFeatureBuilder; import org.geotools.filter.text.cql2.CQLException; import org.geotools.filter.text.ecql.ECQL; +import org.locationtech.geomesa.accumulo.data.AccumuloDataStore; import org.locationtech.geomesa.accumulo.index.Constants; import org.locationtech.geomesa.utils.geotools.SimpleFeatureTypes; import org.opengis.feature.simple.SimpleFeature; @@ -78,7 +80,7 @@ import com.vividsolutions.jts.geom.Geometry; import com.vividsolutions.jts.io.ParseException; import info.aduna.iteration.CloseableIteration; - + /** * A {@link GeoIndexer} wrapper around a GeoMesa {@link AccumuloDataStore}. This class configures and connects to the Datastore, creates the * RDF Feature Type, and interacts with the Datastore. @@ -311,7 +313,7 @@ public class GeoMesaGeoIndexer extends AbstractAccumuloIndexer implements GeoInd } final String filterString = StringUtils.join(filterParms, " AND "); - logger.info("Performing geomesa query : " + filterString); + logger.info("Performing geomesa query : " + LogUtils.clean(filterString)); return getIteratorWrapper(filterString); } @@ -328,7 +330,7 @@ public class GeoMesaGeoIndexer extends AbstractAccumuloIndexer implements GeoInd try { cqlFilter = ECQL.toFilter(filterString); } catch (final CQLException e) { - logger.error("Error parsing query: " + filterString, e); + logger.error("Error parsing query: " + LogUtils.clean(filterString), e); throw new QueryEvaluationException(e); } @@ -336,7 +338,7 @@ public class GeoMesaGeoIndexer extends AbstractAccumuloIndexer implements GeoInd try { featureIterator = featureSource.getFeatures(query).features(); } catch (final IOException e) { - logger.error("Error performing query: " + filterString, e); + logger.error("Error performing query: " + LogUtils.clean(filterString), e); throw new QueryEvaluationException(e); } } @@ -410,7 +412,7 @@ public class GeoMesaGeoIndexer extends AbstractAccumuloIndexer implements GeoInd public CloseableIteration<Statement, QueryEvaluationException> queryOverlaps(final Geometry query, final StatementConstraints contraints) { return performQuery("OVERLAPS", query, contraints); } - + @Override public CloseableIteration<Statement, QueryEvaluationException> queryNear(final NearQuery query, final StatementConstraints contraints) { http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.geoindexing/geo.geowave/src/main/java/org/apache/rya/indexing/accumulo/geo/GeoWaveGeoIndexer.java ---------------------------------------------------------------------- diff --git a/extras/rya.geoindexing/geo.geowave/src/main/java/org/apache/rya/indexing/accumulo/geo/GeoWaveGeoIndexer.java b/extras/rya.geoindexing/geo.geowave/src/main/java/org/apache/rya/indexing/accumulo/geo/GeoWaveGeoIndexer.java index db7af05..0a4e767 100644 --- a/extras/rya.geoindexing/geo.geowave/src/main/java/org/apache/rya/indexing/accumulo/geo/GeoWaveGeoIndexer.java +++ b/extras/rya.geoindexing/geo.geowave/src/main/java/org/apache/rya/indexing/accumulo/geo/GeoWaveGeoIndexer.java @@ -43,6 +43,7 @@ import org.apache.log4j.Logger; import org.apache.rya.accumulo.experimental.AbstractAccumuloIndexer; import org.apache.rya.api.RdfCloudTripleStoreConfiguration; import org.apache.rya.api.domain.RyaStatement; +import org.apache.rya.api.log.LogUtils; import org.apache.rya.api.resolver.RyaToRdfConversions; import org.apache.rya.indexing.GeoIndexer; import org.apache.rya.indexing.Md5Hash; @@ -373,7 +374,7 @@ public class GeoWaveGeoIndexer extends AbstractAccumuloIndexer implements GeoInd } final String filterString = StringUtils.join(filterParms, " AND "); - logger.info("Performing geowave query : " + filterString); + logger.info("Performing geowave query : " + LogUtils.clean(filterString)); return getIteratorWrapper(filterString); } @@ -390,7 +391,7 @@ public class GeoWaveGeoIndexer extends AbstractAccumuloIndexer implements GeoInd try { cqlFilter = ECQL.toFilter(filterString); } catch (final CQLException e) { - logger.error("Error parsing query: " + filterString, e); + logger.error("Error parsing query: " + LogUtils.clean(filterString), e); throw new QueryEvaluationException(e); } @@ -479,7 +480,7 @@ public class GeoWaveGeoIndexer extends AbstractAccumuloIndexer implements GeoInd public CloseableIteration<Statement, QueryEvaluationException> queryOverlaps(final Geometry query, final StatementConstraints contraints) { return performQuery("OVERLAPS", query, contraints); } - + @Override public CloseableIteration<Statement, QueryEvaluationException> queryNear(final NearQuery query, final StatementConstraints contraints) { http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.indexing.pcj/src/main/java/org/apache/rya/indexing/pcj/storage/accumulo/VisibilityBindingSetSerDe.java ---------------------------------------------------------------------- diff --git a/extras/rya.indexing.pcj/src/main/java/org/apache/rya/indexing/pcj/storage/accumulo/VisibilityBindingSetSerDe.java b/extras/rya.indexing.pcj/src/main/java/org/apache/rya/indexing/pcj/storage/accumulo/VisibilityBindingSetSerDe.java index ae43a9a..adb4f27 100644 --- a/extras/rya.indexing.pcj/src/main/java/org/apache/rya/indexing/pcj/storage/accumulo/VisibilityBindingSetSerDe.java +++ b/extras/rya.indexing.pcj/src/main/java/org/apache/rya/indexing/pcj/storage/accumulo/VisibilityBindingSetSerDe.java @@ -22,9 +22,9 @@ import static java.util.Objects.requireNonNull; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; -import java.io.ObjectInputStream; import java.io.ObjectOutputStream; +import org.apache.commons.io.serialization.ValidatingObjectInputStream; import org.apache.fluo.api.data.Bytes; import edu.umd.cs.findbugs.annotations.DefaultAnnotation; @@ -63,9 +63,24 @@ public class VisibilityBindingSetSerDe { */ public VisibilityBindingSet deserialize(final Bytes bytes) throws Exception { requireNonNull(bytes); - - try(final ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes.toArray()))) { - final Object o = ois.readObject(); + try (final ValidatingObjectInputStream vois = new ValidatingObjectInputStream(new ByteArrayInputStream(bytes.toArray()))) { + // Perform input validation. Only the following classes are allowed to be deserialized. + vois.accept(org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSet.class, + org.apache.rya.indexing.pcj.storage.accumulo.BindingSetDecorator.class, + org.openrdf.query.impl.MapBindingSet.class, + java.util.LinkedHashMap.class, + java.util.HashMap.class, + java.math.BigInteger.class, + java.math.BigDecimal.class, + java.lang.Number.class, + org.openrdf.query.impl.BindingImpl.class, + org.openrdf.model.impl.LiteralImpl.class, + org.openrdf.model.impl.IntegerLiteralImpl.class, + org.openrdf.model.impl.DecimalLiteralImpl.class, + org.openrdf.model.impl.URIImpl.class, + org.openrdf.query.algebra.evaluation.QueryBindingSet.class); + vois.accept("[B"); + final Object o = vois.readObject(); if(o instanceof VisibilityBindingSet) { return (VisibilityBindingSet) o; } else { http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/VisibilityBindingSetSerDeTest.java ---------------------------------------------------------------------- diff --git a/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/VisibilityBindingSetSerDeTest.java b/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/VisibilityBindingSetSerDeTest.java index 16f56c1..17042c0 100644 --- a/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/VisibilityBindingSetSerDeTest.java +++ b/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/VisibilityBindingSetSerDeTest.java @@ -21,10 +21,15 @@ package org.apache.rya.indexing.pcj.storage.accumulo; import static org.junit.Assert.assertEquals; +import java.io.ByteArrayOutputStream; +import java.io.InvalidClassException; +import java.io.ObjectOutputStream; +import java.util.ArrayList; + import org.apache.fluo.api.data.Bytes; -import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSet; -import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSetSerDe; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.ExpectedException; import org.openrdf.model.ValueFactory; import org.openrdf.model.impl.ValueFactoryImpl; import org.openrdf.query.impl.MapBindingSet; @@ -49,4 +54,27 @@ public class VisibilityBindingSetSerDeTest { assertEquals(original, result); } + + @Rule + public final ExpectedException exception = ExpectedException.none(); + + /** + * Tests that deserializing an ArrayList should throw an error. + * if VisibilityBindingSet changes to include ArrayList, then this will need changing. + * + * @throws Exception + */ + @Test + public void rejectUnexpectedClass() throws Exception { + // cannot use VisibilityBindingSetSerDe.serialize here since it only serializes VisibilityBindingSet. + final ByteArrayOutputStream boas = new ByteArrayOutputStream(); + try (final ObjectOutputStream oos = new ObjectOutputStream(boas)) { + oos.writeObject(new ArrayList<Integer>()); + } + final Bytes bytes = Bytes.of(boas.toByteArray()); + final VisibilityBindingSetSerDe serde = new VisibilityBindingSetSerDe(); + // Should throw an InvalidClassException when deserializing the wrong class. + exception.expect(InvalidClassException.class); + serde.deserialize(bytes); + } } http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/CopyTool.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/CopyTool.java b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/CopyTool.java index 59b92ba..6fc90ec 100644 --- a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/CopyTool.java +++ b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/CopyTool.java @@ -101,6 +101,7 @@ import org.apache.rya.accumulo.mr.merge.util.ToolConfigUtils; import org.apache.rya.api.RdfCloudTripleStoreConstants; import org.apache.rya.api.RdfCloudTripleStoreUtils; import org.apache.rya.api.layout.TablePrefixLayoutStrategy; +import org.apache.rya.api.path.PathUtils; import org.apache.rya.indexing.accumulo.ConfigUtils; import com.google.common.base.Joiner; @@ -232,23 +233,23 @@ public class CopyTool extends AbstractDualInstanceAccumuloMRTool { public void setup() throws Exception { super.init(); - tempDir = conf.get("hadoop.tmp.dir", null); + tempDir = PathUtils.clean(conf.get("hadoop.tmp.dir", null)); if (tempDir == null) { throw new Exception("Invalid hadoop temp directory. \"hadoop.tmp.dir\" could not be found in the configuration."); } useCopyFileOutput = conf.getBoolean(USE_COPY_FILE_OUTPUT, false); baseOutputDir = tempDir + "/copy_tool_file_output/"; - localBaseOutputDir = conf.get(COPY_FILE_OUTPUT_PATH, null); + localBaseOutputDir = PathUtils.clean(conf.get(COPY_FILE_OUTPUT_PATH, null)); compressionType = conf.get(COPY_FILE_OUTPUT_COMPRESSION_TYPE, null); useCopyFileOutputDirectoryClear = conf.getBoolean(USE_COPY_FILE_OUTPUT_DIRECTORY_CLEAR, false); - localCopyFileImportDir = conf.get(COPY_FILE_IMPORT_DIRECTORY, null); + localCopyFileImportDir = PathUtils.clean(conf.get(COPY_FILE_IMPORT_DIRECTORY, null)); baseImportDir = tempDir + "/copy_tool_import/"; startTime = conf.get(MergeTool.START_TIME_PROP, null); if (!useCopyFileImport) { - if (startTime != null) { + if (startTime != null) { try { final Date date = MergeTool.START_TIME_FORMATTER.parse(startTime); log.info("Will copy all data after " + date); @@ -597,7 +598,7 @@ public class CopyTool extends AbstractDualInstanceAccumuloMRTool { } fs.setPermission(splitsPath, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL)); - final String userDir = System.getProperty("user.dir"); + final String userDir = PathUtils.clean(System.getProperty("user.dir")); // The splits file has a symlink created in the user directory for some reason. // It might be better to copy the entire file for Windows but it doesn't seem to matter if // the user directory symlink is broken. @@ -650,8 +651,11 @@ public class CopyTool extends AbstractDualInstanceAccumuloMRTool { final Path failures = getPath(hdfsBaseWorkDir.toString(), "failures"); // With HDFS permissions on, we need to make sure the Accumulo user can read/move the files - final FsShell shell = new FsShell(conf); - shell.run(new String[] {"-chmod", "777", hdfsBaseWorkDir.toString()}); + final FsShell hdfs = new FsShell(conf); + if (!fs.isDirectory(hdfsBaseWorkDir)) { + throw new IllegalArgumentException("Configured working directory is not a valid directory" + hdfsBaseWorkDir.toString()); + } + hdfs.run(new String[] {"-chmod", "777", hdfsBaseWorkDir.toString()}); if (fs.exists(failures)) { fs.delete(failures, true); } @@ -863,7 +867,7 @@ public class CopyTool extends AbstractDualInstanceAccumuloMRTool { public static void main(final String[] args) { final String log4jConfiguration = System.getProperties().getProperty("log4j.configuration"); if (StringUtils.isNotBlank(log4jConfiguration)) { - final String parsedConfiguration = StringUtils.removeStart(log4jConfiguration, "file:"); + final String parsedConfiguration = PathUtils.clean(StringUtils.removeStart(log4jConfiguration, "file:")); final File configFile = new File(parsedConfiguration); if (configFile.exists()) { DOMConfigurator.configure(parsedConfiguration); http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/MergeTool.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/MergeTool.java b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/MergeTool.java index 21a8316..e52a6cd 100644 --- a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/MergeTool.java +++ b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/MergeTool.java @@ -57,11 +57,6 @@ import org.apache.hadoop.util.ToolRunner; import org.apache.log4j.BasicConfigurator; import org.apache.log4j.Logger; import org.apache.log4j.xml.DOMConfigurator; - -import com.google.common.base.Joiner; -import com.google.common.collect.ImmutableList; -import com.google.common.collect.ImmutableMap; - import org.apache.rya.accumulo.AccumuloRdfConfiguration; import org.apache.rya.accumulo.mr.AccumuloHDFSFileInputFormat; import org.apache.rya.accumulo.mr.MRUtils; @@ -72,8 +67,13 @@ import org.apache.rya.accumulo.mr.merge.util.ToolConfigUtils; import org.apache.rya.api.RdfCloudTripleStoreConfiguration; import org.apache.rya.api.RdfCloudTripleStoreConstants; import org.apache.rya.api.RdfCloudTripleStoreUtils; +import org.apache.rya.api.path.PathUtils; import org.apache.rya.indexing.accumulo.ConfigUtils; +import com.google.common.base.Joiner; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; + /** * Handles merging a child accumulo instance's data back into its parent's * instance. @@ -464,7 +464,7 @@ public class MergeTool extends AbstractDualInstanceAccumuloMRTool { public static void main(final String[] args) { final String log4jConfiguration = System.getProperties().getProperty("log4j.configuration"); if (StringUtils.isNotBlank(log4jConfiguration)) { - final String parsedConfiguration = StringUtils.removeStart(log4jConfiguration, "file:"); + final String parsedConfiguration = PathUtils.clean(StringUtils.removeStart(log4jConfiguration, "file:")); final File configFile = new File(parsedConfiguration); if (configFile.exists()) { DOMConfigurator.configure(parsedConfiguration); http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloInstanceDriver.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloInstanceDriver.java b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloInstanceDriver.java index d3dde34..c600372 100644 --- a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloInstanceDriver.java +++ b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloInstanceDriver.java @@ -48,6 +48,7 @@ import org.apache.rya.accumulo.AccumuloRyaDAO; import org.apache.rya.accumulo.mr.MRUtils; import org.apache.rya.accumulo.mr.merge.MergeTool; import org.apache.rya.api.RdfCloudTripleStoreConstants; +import org.apache.rya.api.path.PathUtils; import org.apache.rya.api.persist.RyaDAOException; import com.google.common.base.Preconditions; @@ -71,7 +72,7 @@ public class AccumuloInstanceDriver { private final boolean isParent; private final String user; - private final String password; + private final String userpwd; private final String instanceName; private final String tablePrefix; private final String auth; @@ -125,18 +126,18 @@ public class AccumuloInstanceDriver { * @param isParent {@code true} if the instance is the parent/main instance. {@code false} if it's the * child. * @param user the user name tied to this instance. - * @param password the password for the user. + * @param userpwd the userpwd for the user. * @param instanceName the name of the instance. * @param tablePrefix the table prefix. * @param auth the comma-separated authorization list. */ - public AccumuloInstanceDriver(final String driverName, final boolean isMock, final boolean shouldCreateIndices, final boolean isReadOnly, final boolean isParent, final String user, final String password, final String instanceName, final String tablePrefix, final String auth) { + public AccumuloInstanceDriver(final String driverName, final boolean isMock, final boolean shouldCreateIndices, final boolean isReadOnly, final boolean isParent, final String user, final String userpwd, final String instanceName, final String tablePrefix, final String auth) { this.driverName = Preconditions.checkNotNull(driverName); this.isMock = isMock; this.shouldCreateIndices = shouldCreateIndices; this.isReadOnly = isReadOnly; this.user = user; - this.password = password; + this.userpwd = userpwd; this.instanceName = instanceName; this.tablePrefix = tablePrefix; this.auth = auth; @@ -166,7 +167,7 @@ public class AccumuloInstanceDriver { // Create and Run MiniAccumulo Cluster tempDir = Files.createTempDir(); tempDir.deleteOnExit(); - miniAccumuloCluster = new MiniAccumuloCluster(tempDir, password); + miniAccumuloCluster = new MiniAccumuloCluster(tempDir, userpwd); copyHadoopHomeToTemp(); miniAccumuloCluster.getConfig().setInstanceName(instanceName); log.info(driverName + " MiniAccumulo instance starting up..."); @@ -176,13 +177,13 @@ public class AccumuloInstanceDriver { log.info("Creating connector to " + driverName + " MiniAccumulo instance..."); zooKeeperInstance = new ZooKeeperInstance(miniAccumuloCluster.getClientConfig()); instance = zooKeeperInstance; - connector = zooKeeperInstance.getConnector(user, new PasswordToken(password)); + connector = zooKeeperInstance.getConnector(user, new PasswordToken(userpwd)); log.info("Created connector to " + driverName + " MiniAccumulo instance"); } else { log.info("Setting up " + driverName + " mock instance..."); mockInstance = new MockInstance(instanceName); instance = mockInstance; - connector = mockInstance.getConnector(user, new PasswordToken(password)); + connector = mockInstance.getConnector(user, new PasswordToken(userpwd)); log.info("Created connector to " + driverName + " mock instance"); } zooKeepers = instance.getZooKeepers(); @@ -196,7 +197,7 @@ public class AccumuloInstanceDriver { */ private void copyHadoopHomeToTemp() throws IOException { if (IS_COPY_HADOOP_HOME_ENABLED && SystemUtils.IS_OS_WINDOWS) { - final String hadoopHomeEnv = System.getenv("HADOOP_HOME"); + final String hadoopHomeEnv = PathUtils.clean(System.getenv("HADOOP_HOME")); if (hadoopHomeEnv != null) { final File hadoopHomeDir = new File(hadoopHomeEnv); if (hadoopHomeDir.exists()) { @@ -255,7 +256,7 @@ public class AccumuloInstanceDriver { log.info("Creating " + driverName + " user and authorizations"); secOps = connector.securityOperations(); if (!user.equals(ROOT_USER_NAME)) { - secOps.createLocalUser(user, new PasswordToken(password)); + secOps.createLocalUser(user, new PasswordToken(userpwd)); } addAuths(auth); final TablePermission tablePermission = isReadOnly ? TablePermission.READ : TablePermission.WRITE; @@ -300,7 +301,7 @@ public class AccumuloInstanceDriver { } configMap.put(MRUtils.AC_INSTANCE_PROP, instanceName); configMap.put(MRUtils.AC_USERNAME_PROP, user); - configMap.put(MRUtils.AC_PWD_PROP, password); + configMap.put(MRUtils.AC_PWD_PROP, userpwd); configMap.put(MRUtils.TABLE_PREFIX_PROPERTY, tablePrefix); configMap.put(MRUtils.AC_AUTH_PROP, auth); configMap.put(MRUtils.AC_ZK_PROP, zooKeepers != null ? zooKeepers : "localhost"); @@ -311,7 +312,9 @@ public class AccumuloInstanceDriver { final String key = entry.getKey(); final String value = entry.getValue(); final String argument = ToolConfigUtils.makeArgument(isParent ? key : key + MergeTool.CHILD_SUFFIX, value); - log.info(argument); + if (!key.equals(MRUtils.AC_PWD_PROP)) { + log.info(argument); + } config.set(key, value); } @@ -448,10 +451,10 @@ public class AccumuloInstanceDriver { } /** - * @return the password for the user. + * @return the userpwd for the user. */ public String getPassword() { - return password; + return userpwd; } /** http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/TimeUtils.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/TimeUtils.java b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/TimeUtils.java index 5c3c72e..ace5c3b 100644 --- a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/TimeUtils.java +++ b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/TimeUtils.java @@ -128,7 +128,10 @@ public final class TimeUtils { } } finally { // Close the connection - conn.disconnect(); + if (conn != null) { + conn.disconnect(); + } + } return remoteDate; http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.merger/src/test/java/org/apache/rya/accumulo/mr/merge/CopyToolTest.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/test/java/org/apache/rya/accumulo/mr/merge/CopyToolTest.java b/extras/rya.merger/src/test/java/org/apache/rya/accumulo/mr/merge/CopyToolTest.java index 59f75a5..fcf4d9e 100644 --- a/extras/rya.merger/src/test/java/org/apache/rya/accumulo/mr/merge/CopyToolTest.java +++ b/extras/rya.merger/src/test/java/org/apache/rya/accumulo/mr/merge/CopyToolTest.java @@ -26,6 +26,7 @@ import static org.apache.rya.accumulo.mr.merge.util.ToolConfigUtils.makeArgument import static org.junit.Assert.assertEquals; import static org.junit.Assert.fail; +import java.io.File; import java.util.Date; import java.util.Iterator; import java.util.List; @@ -43,13 +44,6 @@ import org.apache.accumulo.core.security.Authorizations; import org.apache.accumulo.core.security.ColumnVisibility; import org.apache.hadoop.conf.Configuration; import org.apache.log4j.Logger; -import org.junit.After; -import org.junit.AfterClass; -import org.junit.Before; -import org.junit.BeforeClass; -import org.junit.Test; - -import info.aduna.iteration.CloseableIteration; import org.apache.rya.accumulo.AccumuloRdfConfiguration; import org.apache.rya.accumulo.AccumuloRyaDAO; import org.apache.rya.accumulo.mr.MRUtils; @@ -63,6 +57,14 @@ import org.apache.rya.api.RdfCloudTripleStoreConstants; import org.apache.rya.api.domain.RyaStatement; import org.apache.rya.api.persist.RyaDAOException; import org.apache.rya.indexing.accumulo.ConfigUtils; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import info.aduna.iteration.CloseableIteration; /** * Tests for {@link CopyTool}. @@ -98,6 +100,25 @@ public class CopyToolTest { private static CopyTool copyTool = null; private boolean isImporting = false; + public static String getProjectRootDir() { + String rootDir = System.getProperty("basedir"); + if(rootDir == null) { + rootDir = System.getProperty("user.dir"); + } + if(rootDir == null) { + throw new RuntimeException("Expected user.dir to contain a value"); + } + return rootDir; + } + + private static File getUnitTestScratchDirectory(final String testName) { + final File dir = new File(getProjectRootDir() + File.separator + "target" + + File.separator + "TestScratch" + File.separator + + testName+ "-" + System.currentTimeMillis()); + Assert.assertTrue("Unable to make TestScratchDirectory:"+ dir.getAbsolutePath(), dir.mkdirs()); + return dir; + } + @BeforeClass public static void setUp() throws Exception { accumuloDualInstanceDriver = new AccumuloDualInstanceDriver(IS_MOCK, true, true, false, false); @@ -161,12 +182,13 @@ public class CopyToolTest { makeArgument(CopyTool.NTP_SERVER_HOST_PROP, TimeUtils.DEFAULT_TIME_SERVER_HOST), makeArgument(CopyTool.USE_NTP_SERVER_PROP, Boolean.toString(USE_TIME_SYNC)), makeArgument(CopyTool.USE_COPY_FILE_OUTPUT, Boolean.toString(USE_COPY_FILE_OUTPUT)), - makeArgument(CopyTool.COPY_FILE_OUTPUT_PATH, "/test/copy_tool_file_output/"), + makeArgument(CopyTool.COPY_FILE_OUTPUT_PATH, getUnitTestScratchDirectory(CopyToolTest.class.getSimpleName() + "-copyFileOutput").getAbsolutePath()), makeArgument(CopyTool.COPY_FILE_OUTPUT_COMPRESSION_TYPE, Algorithm.GZ.getName()), makeArgument(CopyTool.USE_COPY_FILE_OUTPUT_DIRECTORY_CLEAR, Boolean.toString(true)), makeArgument(CopyTool.COPY_FILE_IMPORT_DIRECTORY, "resources/test/copy_tool_file_output/"), makeArgument(CopyTool.USE_COPY_FILE_IMPORT, Boolean.toString(isImporting)), - makeArgument(MergeTool.START_TIME_PROP, MergeTool.getStartTimeString(startDate)) + makeArgument(MergeTool.START_TIME_PROP, MergeTool.getStartTimeString(startDate)), + makeArgument("hadoop.tmp.dir", getUnitTestScratchDirectory(CopyToolTest.class.getSimpleName()).getAbsolutePath()) }); final Configuration toolConfig = copyTool.getConf(); http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.merger/src/test/java/org/apache/rya/accumulo/mr/merge/RulesetCopyIT.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/test/java/org/apache/rya/accumulo/mr/merge/RulesetCopyIT.java b/extras/rya.merger/src/test/java/org/apache/rya/accumulo/mr/merge/RulesetCopyIT.java index b85ee4e..1c83378 100644 --- a/extras/rya.merger/src/test/java/org/apache/rya/accumulo/mr/merge/RulesetCopyIT.java +++ b/extras/rya.merger/src/test/java/org/apache/rya/accumulo/mr/merge/RulesetCopyIT.java @@ -21,6 +21,7 @@ package org.apache.rya.accumulo.mr.merge; import static org.apache.rya.accumulo.mr.merge.util.TestUtils.YESTERDAY; import static org.apache.rya.accumulo.mr.merge.util.ToolConfigUtils.makeArgument; +import java.io.File; import java.util.Arrays; import java.util.HashMap; import java.util.HashSet; @@ -144,6 +145,25 @@ public class RulesetCopyIT { return new RyaType(type, lit); } + public static String getProjectRootDir() { + String rootDir = System.getProperty("basedir"); + if(rootDir == null) { + rootDir = System.getProperty("user.dir"); + } + if(rootDir == null) { + throw new RuntimeException("Expected user.dir to contain a value"); + } + return rootDir; + } + + private static File getUnitTestScratchDirectory(final String testName) { + final File dir = new File(getProjectRootDir() + File.separator + "target" + + File.separator + "TestScratch" + File.separator + + testName+ "-" + System.currentTimeMillis()); + Assert.assertTrue("Unable to make TestScratchDirectory:"+ dir.getAbsolutePath(), dir.mkdirs()); + return dir; + } + @BeforeClass public static void setUpPerClass() throws Exception { DemoUtilities.setupLogging(LoggingDetail.LIGHT); @@ -212,7 +232,8 @@ public class RulesetCopyIT { makeArgument(CopyTool.CREATE_CHILD_INSTANCE_TYPE_PROP, (IS_MOCK ? InstanceType.MOCK : InstanceType.MINI).toString()), makeArgument(CopyTool.QUERY_STRING_PROP, query), makeArgument(CopyTool.USE_COPY_QUERY_SPARQL, "true"), - makeArgument(RdfCloudTripleStoreConfiguration.CONF_INFER, Boolean.toString(infer)) + makeArgument(RdfCloudTripleStoreConfiguration.CONF_INFER, Boolean.toString(infer)), + makeArgument("hadoop.tmp.dir", getUnitTestScratchDirectory(RulesetCopyIT.class.getSimpleName()).getAbsolutePath()) }); final Configuration toolConfig = rulesetTool.getConf(); http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/AggregationResultUpdater.java ---------------------------------------------------------------------- diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/AggregationResultUpdater.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/AggregationResultUpdater.java index 3acf3e9..e79528c 100644 --- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/AggregationResultUpdater.java +++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/AggregationResultUpdater.java @@ -24,7 +24,6 @@ import static java.util.Objects.requireNonNull; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; -import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.io.Serializable; import java.math.BigDecimal; @@ -34,10 +33,12 @@ import java.util.Map; import java.util.Objects; import java.util.Optional; +import org.apache.commons.io.serialization.ValidatingObjectInputStream; import org.apache.fluo.api.client.TransactionBase; import org.apache.fluo.api.data.Bytes; import org.apache.log4j.Logger; import org.apache.rya.accumulo.utils.VisibilitySimplifier; +import org.apache.rya.api.log.LogUtils; import org.apache.rya.indexing.pcj.fluo.app.query.AggregationMetadata; import org.apache.rya.indexing.pcj.fluo.app.query.AggregationMetadata.AggregationElement; import org.apache.rya.indexing.pcj.fluo.app.query.AggregationMetadata.AggregationType; @@ -126,7 +127,7 @@ public class AggregationResultUpdater extends AbstractNodeUpdater { log.trace( "Transaction ID: " + tx.getStartTimestamp() + "\n" + - "Before Update: " + state.getBindingSet().toString() + "\n"); + "Before Update: " + LogUtils.clean(state.getBindingSet().toString()) + "\n"); // Update the visibilities of the result binding set based on the child's visibilities. final String oldVisibility = state.getVisibility(); @@ -146,7 +147,7 @@ public class AggregationResultUpdater extends AbstractNodeUpdater { log.trace( "Transaction ID: " + tx.getStartTimestamp() + "\n" + - "After Update:" + state.getBindingSet().toString() + "\n" ); + "After Update:" + LogUtils.clean(state.getBindingSet().toString()) + "\n" ); // Store the updated state. This will write on top of any old state that was present for the Group By values. tx.set(rowId, FluoQueryColumns.AGGREGATION_BINDING_SET, Bytes.of(AGG_STATE_SERDE.serialize(state))); @@ -403,8 +404,27 @@ public class AggregationResultUpdater extends AbstractNodeUpdater { final AggregationState state; final ByteArrayInputStream bais = new ByteArrayInputStream(bytes); - try(ObjectInputStream ois = new ObjectInputStream(bais)) { - final Object o = ois.readObject(); + try(ValidatingObjectInputStream vois = new ValidatingObjectInputStream(bais) + //// this is how you find classes that you missed in the vois.accept() list, below. + // { @Override protected void invalidClassNameFound(String className) throws java.io.InvalidClassException { + // System.out.println("vois.accept(" + className + ".class, ");};}; + ) { + // These classes are allowed to be deserialized. Others throw InvalidClassException. + vois.accept(org.apache.rya.indexing.pcj.fluo.app.AggregationResultUpdater.AggregationState.class, // + org.apache.rya.indexing.pcj.fluo.app.AggregationResultUpdater.AverageState.class, // + java.util.HashMap.class, // + java.math.BigInteger.class, // + java.lang.Number.class, // + java.math.BigDecimal.class, // + org.openrdf.query.impl.MapBindingSet.class, // + java.util.LinkedHashMap.class, // + org.openrdf.query.impl.BindingImpl.class, // + org.openrdf.model.impl.URIImpl.class, // + org.openrdf.model.impl.LiteralImpl.class, // + org.openrdf.model.impl.DecimalLiteralImpl.class, // + org.openrdf.model.impl.IntegerLiteralImpl.class); + vois.accept("[B"); // Array of Bytes + final Object o = vois.readObject(); if(o instanceof AggregationState) { state = (AggregationState)o; } else { http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/kafka/KafkaBindingSetExporter.java ---------------------------------------------------------------------- diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/kafka/KafkaBindingSetExporter.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/kafka/KafkaBindingSetExporter.java index 61b3aa2..f7e3489 100644 --- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/kafka/KafkaBindingSetExporter.java +++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/kafka/KafkaBindingSetExporter.java @@ -29,6 +29,7 @@ import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.rya.api.client.CreatePCJ.ExportStrategy; import org.apache.rya.api.client.CreatePCJ.QueryType; +import org.apache.rya.api.log.LogUtils; import org.apache.rya.indexing.pcj.fluo.app.export.IncrementalBindingSetExporter; import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSet; import org.slf4j.Logger; @@ -71,9 +72,9 @@ public class KafkaBindingSetExporter implements IncrementalBindingSetExporter { // Don't let the export return until the result has been written to the topic. Otherwise we may lose results. future.get(); - - log.debug("Producer successfully sent record with queryId: {} and visbilityBindingSet: \n{}", queryId, result); - + if(log.isDebugEnabled()) { + log.debug("Producer successfully sent record with queryId: {} and visbilityBindingSet: \n{}", LogUtils.clean(queryId), LogUtils.clean(result.toString())); + } } catch (final Throwable e) { throw new ResultExportException("A result could not be exported to Kafka.", e); } http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/7b571d43/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataDAO.java ---------------------------------------------------------------------- diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataDAO.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataDAO.java index 55e521e..ba75a56 100644 --- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataDAO.java +++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataDAO.java @@ -22,7 +22,7 @@ import static java.util.Objects.requireNonNull; import java.io.ByteArrayOutputStream; import java.io.IOException; -import java.io.ObjectInputStream; +import java.io.InvalidClassException; import java.io.ObjectOutputStream; import java.util.Collection; import java.util.HashSet; @@ -30,6 +30,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; +import org.apache.commons.io.serialization.ValidatingObjectInputStream; import org.apache.fluo.api.client.SnapshotBase; import org.apache.fluo.api.client.TransactionBase; import org.apache.fluo.api.data.Bytes; @@ -560,8 +561,21 @@ public class FluoQueryMetadataDAO { // Deserialize the collection of AggregationElements. final Bytes aggBytes = sx.get(Bytes.of(nodeId.getBytes(Charsets.UTF_8)), FluoQueryColumns.AGGREGATION_AGGREGATIONS); final Collection<AggregationElement> aggregations; - try(final ObjectInputStream ois = new ObjectInputStream(aggBytes.toInputStream())) { - aggregations = (Collection<AggregationElement>)ois.readObject(); + try (final ValidatingObjectInputStream vois = new ValidatingObjectInputStream(aggBytes.toInputStream()) + //// this is how you find classes that you missed in the vois.accept() list, below. + // { @Override protected void invalidClassNameFound(String className) throws java.io.InvalidClassException { + // System.out.println("vois.accept(" + className + ".class, ");};}; + ) { + // These classes are allowed to be deserialized. Others throw InvalidClassException. + vois.accept(java.util.ArrayList.class, // + java.lang.Enum.class, // + org.apache.rya.indexing.pcj.fluo.app.query.AggregationMetadata.AggregationElement.class, // + org.apache.rya.indexing.pcj.fluo.app.query.AggregationMetadata.AggregationType.class); + final Object object = vois.readObject(); + if (!(object instanceof Collection<?>)) { + throw new InvalidClassException("Object read was not of type Collection. It was: " + object.getClass()); + } + aggregations = (Collection<AggregationElement>) object; } catch (final IOException | ClassNotFoundException e) { throw new RuntimeException("Problem encountered while reading AggregationMetadata from the Fluo table. Unable " + "to deserialize the AggregationElements from a byte[].", e);
