RYA-377 Adding a Kafka Streams Processor that outputs VisibilityStatements from VisibilityBindingSets.
Project: http://git-wip-us.apache.org/repos/asf/incubator-rya/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-rya/commit/b8b0a126 Tree: http://git-wip-us.apache.org/repos/asf/incubator-rya/tree/b8b0a126 Diff: http://git-wip-us.apache.org/repos/asf/incubator-rya/diff/b8b0a126 Branch: refs/heads/master Commit: b8b0a1269d7b2ae195dab01d4305b36c258bd2e4 Parents: 98af7aa Author: kchilton2 <[email protected]> Authored: Wed Nov 15 18:18:08 2017 -0500 Committer: caleb <[email protected]> Committed: Tue Jan 9 15:13:00 2018 -0500 ---------------------------------------------------------------------- extras/rya.streams/kafka/pom.xml | 5 + .../RyaStreamsSinkFormatterSupplier.java | 83 -------- .../BindingSetOutputFormatterSupplier.java | 84 ++++++++ .../StatementOutputFormatterSupplier.java | 114 +++++++++++ .../kafka/processors/ProjectionProcessorIT.java | 6 +- .../processors/StatementPatternProcessorIT.java | 10 +- .../kafka/processors/join/JoinProcessorIT.java | 12 +- .../output/BindingSetOutputFormatterTest.java | 84 ++++++++ .../output/StatementOutputFormatterTest.java | 203 +++++++++++++++++++ 9 files changed, 504 insertions(+), 97 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/b8b0a126/extras/rya.streams/kafka/pom.xml ---------------------------------------------------------------------- diff --git a/extras/rya.streams/kafka/pom.xml b/extras/rya.streams/kafka/pom.xml index d3f6891..16a8b8e 100644 --- a/extras/rya.streams/kafka/pom.xml +++ b/extras/rya.streams/kafka/pom.xml @@ -79,5 +79,10 @@ under the License. <artifactId>rya.test.kafka</artifactId> <scope>test</scope> </dependency> + <dependency> + <groupId>org.mockito</groupId> + <artifactId>mockito-all</artifactId> + <scope>test</scope> + </dependency> </dependencies> </project> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/b8b0a126/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/processors/RyaStreamsSinkFormatterSupplier.java ---------------------------------------------------------------------- diff --git a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/processors/RyaStreamsSinkFormatterSupplier.java b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/processors/RyaStreamsSinkFormatterSupplier.java deleted file mode 100644 index d6a8d2d..0000000 --- a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/processors/RyaStreamsSinkFormatterSupplier.java +++ /dev/null @@ -1,83 +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.rya.streams.kafka.processors; - -import org.apache.kafka.streams.processor.Processor; -import org.apache.kafka.streams.processor.ProcessorContext; -import org.apache.kafka.streams.processor.ProcessorSupplier; -import org.apache.rya.api.model.VisibilityBindingSet; - -import edu.umd.cs.findbugs.annotations.DefaultAnnotation; -import edu.umd.cs.findbugs.annotations.NonNull; - -/** - * Supplies {@link RyaStreamsSinkFormatter} instances. - */ -@DefaultAnnotation(NonNull.class) -public class RyaStreamsSinkFormatterSupplier implements ProcessorSupplier<Object, ProcessorResult> { - - @Override - public Processor<Object, ProcessorResult> get() { - return new RyaStreamsSinkFormatter(); - } - - /** - * Accepts {@link ProcessorResult}s and forwards just their {@link VisibilityBindingSet} so that it may be - * written to a sink. - */ - @DefaultAnnotation(NonNull.class) - public static final class RyaStreamsSinkFormatter implements Processor<Object, ProcessorResult> { - - private ProcessorContext processorContext; - - @Override - public void init(final ProcessorContext context) { - processorContext = context; - } - - @Override - public void process(final Object key, final ProcessorResult value) { - - VisibilityBindingSet result = null; - switch(value.getType()) { - case UNARY: - result = value.getUnary().getResult(); - break; - - case BINARY: - result = value.getBinary().getResult(); - break; - } - - if(result != null) { - processorContext.forward(key, result); - } - } - - @Override - public void punctuate(final long timestamp) { - // Does nothing. - } - - @Override - public void close() { - // Does nothing. - } - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/b8b0a126/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/processors/output/BindingSetOutputFormatterSupplier.java ---------------------------------------------------------------------- diff --git a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/processors/output/BindingSetOutputFormatterSupplier.java b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/processors/output/BindingSetOutputFormatterSupplier.java new file mode 100644 index 0000000..89f5c6c --- /dev/null +++ b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/processors/output/BindingSetOutputFormatterSupplier.java @@ -0,0 +1,84 @@ +/* + * 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.streams.kafka.processors.output; + +import org.apache.kafka.streams.processor.Processor; +import org.apache.kafka.streams.processor.ProcessorContext; +import org.apache.kafka.streams.processor.ProcessorSupplier; +import org.apache.rya.api.model.VisibilityBindingSet; +import org.apache.rya.streams.kafka.processors.ProcessorResult; + +import edu.umd.cs.findbugs.annotations.DefaultAnnotation; +import edu.umd.cs.findbugs.annotations.NonNull; + +/** + * Supplies {@link BindingSetOutputFormatter} instances. + */ +@DefaultAnnotation(NonNull.class) +public class BindingSetOutputFormatterSupplier implements ProcessorSupplier<Object, ProcessorResult> { + + @Override + public Processor<Object, ProcessorResult> get() { + return new BindingSetOutputFormatter(); + } + + /** + * Accepts {@link ProcessorResult}s and forwards just their {@link VisibilityBindingSet} so that it may be + * written to a sink. + */ + @DefaultAnnotation(NonNull.class) + public static final class BindingSetOutputFormatter implements Processor<Object, ProcessorResult> { + + private ProcessorContext processorContext; + + @Override + public void init(final ProcessorContext context) { + processorContext = context; + } + + @Override + public void process(final Object key, final ProcessorResult value) { + + VisibilityBindingSet result = null; + switch(value.getType()) { + case UNARY: + result = value.getUnary().getResult(); + break; + + case BINARY: + result = value.getBinary().getResult(); + break; + } + + if(result != null) { + processorContext.forward(key, result); + } + } + + @Override + public void punctuate(final long timestamp) { + // Does nothing. + } + + @Override + public void close() { + // Does nothing. + } + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/b8b0a126/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/processors/output/StatementOutputFormatterSupplier.java ---------------------------------------------------------------------- diff --git a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/processors/output/StatementOutputFormatterSupplier.java b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/processors/output/StatementOutputFormatterSupplier.java new file mode 100644 index 0000000..74a6f96 --- /dev/null +++ b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/processors/output/StatementOutputFormatterSupplier.java @@ -0,0 +1,114 @@ +/* + * 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.streams.kafka.processors.output; + +import java.util.Collection; + +import org.apache.kafka.streams.processor.Processor; +import org.apache.kafka.streams.processor.ProcessorContext; +import org.apache.kafka.streams.processor.ProcessorSupplier; +import org.apache.rya.api.model.VisibilityBindingSet; +import org.apache.rya.api.model.VisibilityStatement; +import org.apache.rya.streams.kafka.processors.ProcessorResult; +import org.openrdf.model.Resource; +import org.openrdf.model.Statement; +import org.openrdf.model.URI; +import org.openrdf.model.Value; +import org.openrdf.model.ValueFactory; +import org.openrdf.model.impl.ValueFactoryImpl; + +import com.google.common.collect.Sets; + +import edu.umd.cs.findbugs.annotations.DefaultAnnotation; +import edu.umd.cs.findbugs.annotations.NonNull; + +/** + * Supplies instance of {@link StatementOutputFormatter} + */ +@DefaultAnnotation(NonNull.class) +public class StatementOutputFormatterSupplier implements ProcessorSupplier<Object, ProcessorResult> { + + @Override + public Processor<Object, ProcessorResult> get() { + return new StatementOutputFormatter(); + } + + /** + * Converts {@link VisiblityBindingSet}s that contain a "subject", "predicate", and "object" binding into a + * {@link VisibilityStatement} and then forwards that to the downstream processor. + */ + @DefaultAnnotation(NonNull.class) + public static final class StatementOutputFormatter implements Processor<Object, ProcessorResult> { + + private static final ValueFactory VF = new ValueFactoryImpl(); + private static final Collection<String> REQURIED_BINDINGS = Sets.newHashSet("subject", "predicate", "object"); + + private ProcessorContext processorContext; + + @Override + public void init(final ProcessorContext context) { + processorContext = context; + } + + @Override + public void process(final Object key, final ProcessorResult value) { + VisibilityBindingSet result = null; + switch(value.getType()) { + case UNARY: + result = value.getUnary().getResult(); + break; + + case BINARY: + result = value.getBinary().getResult(); + break; + } + + if(result != null && result.getBindingNames().containsAll(REQURIED_BINDINGS)) { + // Make sure the Subject is the correct type. + final Value subjVal = result.getValue("subject"); + if(!(subjVal instanceof Resource)) { + return; + } + + // Make sure the Predicate is the correct type. + final Value predVal = result.getValue("predicate"); + if(!(predVal instanceof URI)) { + return; + } + + // Forward the visibility statement. + final Statement statement = VF.createStatement( + (Resource) subjVal, + (URI) predVal, + result.getValue("object")); + processorContext.forward(key, new VisibilityStatement(statement, result.getVisibility())); + } + } + + @Override + public void punctuate(final long timestamp) { + // Do nothing. + } + + @Override + public void close() { + // Do nothing. + } + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/b8b0a126/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/ProjectionProcessorIT.java ---------------------------------------------------------------------- diff --git a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/ProjectionProcessorIT.java b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/ProjectionProcessorIT.java index 53d1765..f58387e 100644 --- a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/ProjectionProcessorIT.java +++ b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/ProjectionProcessorIT.java @@ -33,8 +33,8 @@ import org.apache.rya.streams.kafka.KafkaTestUtil; import org.apache.rya.streams.kafka.KafkaTopics; import org.apache.rya.streams.kafka.RdfTestUtil; import org.apache.rya.streams.kafka.processors.ProcessorResult.UnaryResult; -import org.apache.rya.streams.kafka.processors.RyaStreamsSinkFormatterSupplier.RyaStreamsSinkFormatter; import org.apache.rya.streams.kafka.processors.StatementPatternProcessorSupplier.StatementPatternProcessor; +import org.apache.rya.streams.kafka.processors.output.BindingSetOutputFormatterSupplier.BindingSetOutputFormatter; import org.apache.rya.streams.kafka.processors.projection.ProjectionProcessorSupplier; import org.apache.rya.streams.kafka.serialization.VisibilityBindingSetSerializer; import org.apache.rya.streams.kafka.serialization.VisibilityStatementDeserializer; @@ -85,7 +85,7 @@ public class ProjectionProcessorIT { builder.addProcessor("P1", new ProjectionProcessorSupplier(elems, result -> ProcessorResult.make(new UnaryResult(result))), "SP1"); // Add a processor that formats the VisibilityBindingSet for output. - builder.addProcessor("SINK_FORMATTER", RyaStreamsSinkFormatter::new, "P1"); + builder.addProcessor("SINK_FORMATTER", BindingSetOutputFormatter::new, "P1"); // Add a sink that writes the data out to a new Kafka topic. builder.addSink("QUERY_RESULTS", resultsTopic, new StringSerializer(), new VisibilityBindingSetSerializer(), "SINK_FORMATTER"); @@ -131,7 +131,7 @@ public class ProjectionProcessorIT { builder.addProcessor("P1", new ProjectionProcessorSupplier(elems, result -> ProcessorResult.make(new UnaryResult(result))), "SP1"); // Add a processor that formats the VisibilityBindingSet for output. - builder.addProcessor("SINK_FORMATTER", RyaStreamsSinkFormatter::new, "P1"); + builder.addProcessor("SINK_FORMATTER", BindingSetOutputFormatter::new, "P1"); // Add a sink that writes the data out to a new Kafka topic. builder.addSink("QUERY_RESULTS", resultsTopic, new StringSerializer(), new VisibilityBindingSetSerializer(), "SINK_FORMATTER"); http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/b8b0a126/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/StatementPatternProcessorIT.java ---------------------------------------------------------------------- diff --git a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/StatementPatternProcessorIT.java b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/StatementPatternProcessorIT.java index 371fd0b..0b2ff60 100644 --- a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/StatementPatternProcessorIT.java +++ b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/StatementPatternProcessorIT.java @@ -33,8 +33,8 @@ import org.apache.rya.streams.kafka.KafkaTestUtil; import org.apache.rya.streams.kafka.KafkaTopics; import org.apache.rya.streams.kafka.RdfTestUtil; import org.apache.rya.streams.kafka.processors.ProcessorResult.UnaryResult; -import org.apache.rya.streams.kafka.processors.RyaStreamsSinkFormatterSupplier.RyaStreamsSinkFormatter; import org.apache.rya.streams.kafka.processors.StatementPatternProcessorSupplier.StatementPatternProcessor; +import org.apache.rya.streams.kafka.processors.output.BindingSetOutputFormatterSupplier.BindingSetOutputFormatter; import org.apache.rya.streams.kafka.serialization.VisibilityBindingSetSerializer; import org.apache.rya.streams.kafka.serialization.VisibilityStatementDeserializer; import org.apache.rya.test.kafka.KafkaTestInstanceRule; @@ -74,7 +74,7 @@ public class StatementPatternProcessorIT { builder.addProcessor("SP1", new StatementPatternProcessorSupplier(sp, result -> ProcessorResult.make( new UnaryResult(result) )), "STATEMENTS"); // Add a processor that formats the VisibilityBindingSet for output. - builder.addProcessor("SINK_FORMATTER", RyaStreamsSinkFormatter::new, "SP1"); + builder.addProcessor("SINK_FORMATTER", BindingSetOutputFormatter::new, "SP1"); // Add a sink that writes the data out to a new Kafka topic. builder.addSink("QUERY_RESULTS", resultsTopic, new StringSerializer(), new VisibilityBindingSetSerializer(), "SINK_FORMATTER"); @@ -117,7 +117,7 @@ public class StatementPatternProcessorIT { builder.addProcessor("SP1", new StatementPatternProcessorSupplier(sp, result -> ProcessorResult.make( new UnaryResult(result) )), "STATEMENTS"); // Add a processor that formats the VisibilityBindingSet for output. - builder.addProcessor("SINK_FORMATTER", RyaStreamsSinkFormatter::new, "SP1"); + builder.addProcessor("SINK_FORMATTER", BindingSetOutputFormatter::new, "SP1"); // Add a sink that writes the data out to a new Kafka topic. builder.addSink("QUERY_RESULTS", resultsTopic, new StringSerializer(), new VisibilityBindingSetSerializer(), "SINK_FORMATTER"); @@ -172,7 +172,7 @@ public class StatementPatternProcessorIT { builder.addProcessor("SP2", new StatementPatternProcessorSupplier(sp2, result -> ProcessorResult.make( new UnaryResult(result) )), "STATEMENTS"); // Add a processor that formats the VisibilityBindingSet for output. - builder.addProcessor("SINK_FORMATTER", RyaStreamsSinkFormatter::new, "SP1", "SP2"); + builder.addProcessor("SINK_FORMATTER", BindingSetOutputFormatter::new, "SP1", "SP2"); // Add a sink that writes the data out to a new Kafka topic. builder.addSink("QUERY_RESULTS", resultsTopic, new StringSerializer(), new VisibilityBindingSetSerializer(), "SINK_FORMATTER"); @@ -224,7 +224,7 @@ public class StatementPatternProcessorIT { builder.addProcessor("SP2", new StatementPatternProcessorSupplier(sp2, result -> ProcessorResult.make( new UnaryResult(result) )), "STATEMENTS"); // Add a processor that formats the VisibilityBindingSet for output. - builder.addProcessor("SINK_FORMATTER", RyaStreamsSinkFormatter::new, "SP1", "SP2"); + builder.addProcessor("SINK_FORMATTER", BindingSetOutputFormatter::new, "SP1", "SP2"); // Add a sink that writes the data out to a new Kafka topic. builder.addSink("QUERY_RESULTS", resultsTopic, new StringSerializer(), new VisibilityBindingSetSerializer(), "SINK_FORMATTER"); http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/b8b0a126/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/join/JoinProcessorIT.java ---------------------------------------------------------------------- diff --git a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/join/JoinProcessorIT.java b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/join/JoinProcessorIT.java index 14a559f..7051efa 100644 --- a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/join/JoinProcessorIT.java +++ b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/join/JoinProcessorIT.java @@ -40,9 +40,9 @@ import org.apache.rya.streams.kafka.processors.ProcessorResult; import org.apache.rya.streams.kafka.processors.ProcessorResult.BinaryResult; import org.apache.rya.streams.kafka.processors.ProcessorResult.BinaryResult.Side; import org.apache.rya.streams.kafka.processors.ProcessorResult.UnaryResult; -import org.apache.rya.streams.kafka.processors.RyaStreamsSinkFormatterSupplier.RyaStreamsSinkFormatter; import org.apache.rya.streams.kafka.processors.StatementPatternProcessorSupplier; import org.apache.rya.streams.kafka.processors.join.JoinProcessorSupplier.JoinProcessor; +import org.apache.rya.streams.kafka.processors.output.BindingSetOutputFormatterSupplier.BindingSetOutputFormatter; import org.apache.rya.streams.kafka.serialization.VisibilityBindingSetSerde; import org.apache.rya.streams.kafka.serialization.VisibilityBindingSetSerializer; import org.apache.rya.streams.kafka.serialization.VisibilityStatementDeserializer; @@ -118,7 +118,7 @@ public class JoinProcessorIT { builder.addStateStore(joinStoreSupplier, "NATURAL_JOIN"); // Add a processor that formats the VisibilityBindingSet for output. - builder.addProcessor("SINK_FORMATTER", RyaStreamsSinkFormatter::new, "NATURAL_JOIN"); + builder.addProcessor("SINK_FORMATTER", BindingSetOutputFormatter::new, "NATURAL_JOIN"); // Add a sink that writes the data out to a new Kafka topic. builder.addSink("QUERY_RESULTS", resultsTopic, new StringSerializer(), new VisibilityBindingSetSerializer(), "SINK_FORMATTER"); @@ -201,7 +201,7 @@ public class JoinProcessorIT { builder.addStateStore(joinStoreSupplier, "NATURAL_JOIN"); // Add a processor that formats the VisibilityBindingSet for output. - builder.addProcessor("SINK_FORMATTER", RyaStreamsSinkFormatter::new, "NATURAL_JOIN"); + builder.addProcessor("SINK_FORMATTER", BindingSetOutputFormatter::new, "NATURAL_JOIN"); // Add a sink that writes the data out to a new Kafka topic. builder.addSink("QUERY_RESULTS", resultsTopic, new StringSerializer(), new VisibilityBindingSetSerializer(), "SINK_FORMATTER"); @@ -284,7 +284,7 @@ public class JoinProcessorIT { builder.addStateStore(joinStoreSupplier, "NATURAL_JOIN"); // Add a processor that formats the VisibilityBindingSet for output. - builder.addProcessor("SINK_FORMATTER", RyaStreamsSinkFormatter::new, "NATURAL_JOIN"); + builder.addProcessor("SINK_FORMATTER", BindingSetOutputFormatter::new, "NATURAL_JOIN"); // Add a sink that writes the data out to a new Kafka topic. builder.addSink("QUERY_RESULTS", resultsTopic, new StringSerializer(), new VisibilityBindingSetSerializer(), "SINK_FORMATTER"); @@ -394,7 +394,7 @@ public class JoinProcessorIT { builder.addStateStore(join2StoreSupplier, "JOIN2"); // Add a processor that formats the VisibilityBindingSet for output. - builder.addProcessor("SINK_FORMATTER", RyaStreamsSinkFormatter::new, "JOIN2"); + builder.addProcessor("SINK_FORMATTER", BindingSetOutputFormatter::new, "JOIN2"); // Add a sink that writes the data out to a new Kafka topic. builder.addSink("QUERY_RESULTS", resultsTopic, new StringSerializer(), new VisibilityBindingSetSerializer(), "SINK_FORMATTER"); @@ -466,7 +466,7 @@ public class JoinProcessorIT { builder.addStateStore(joinStoreSupplier, "LEFT_JOIN"); // Add a processor that formats the VisibilityBindingSet for output. - builder.addProcessor("SINK_FORMATTER", RyaStreamsSinkFormatter::new, "LEFT_JOIN"); + builder.addProcessor("SINK_FORMATTER", BindingSetOutputFormatter::new, "LEFT_JOIN"); // Add a sink that writes the data out to a new Kafka topic. builder.addSink("QUERY_RESULTS", resultsTopic, new StringSerializer(), new VisibilityBindingSetSerializer(), "SINK_FORMATTER"); http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/b8b0a126/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/output/BindingSetOutputFormatterTest.java ---------------------------------------------------------------------- diff --git a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/output/BindingSetOutputFormatterTest.java b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/output/BindingSetOutputFormatterTest.java new file mode 100644 index 0000000..3c810b1 --- /dev/null +++ b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/output/BindingSetOutputFormatterTest.java @@ -0,0 +1,84 @@ +/* + * 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.streams.kafka.processors.output; + +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import org.apache.kafka.streams.processor.ProcessorContext; +import org.apache.rya.api.model.VisibilityBindingSet; +import org.apache.rya.streams.kafka.processors.ProcessorResult; +import org.apache.rya.streams.kafka.processors.ProcessorResult.BinaryResult; +import org.apache.rya.streams.kafka.processors.ProcessorResult.BinaryResult.Side; +import org.apache.rya.streams.kafka.processors.ProcessorResult.UnaryResult; +import org.apache.rya.streams.kafka.processors.output.BindingSetOutputFormatterSupplier.BindingSetOutputFormatter; +import org.junit.Test; +import org.openrdf.model.ValueFactory; +import org.openrdf.model.impl.ValueFactoryImpl; +import org.openrdf.query.impl.MapBindingSet; + +/** + * Unit tests the methods of {@link BindingSetOutputFormatter}. + */ +public class BindingSetOutputFormatterTest { + + @Test + public void unaryResult() { + // Create the input binding set. + final ValueFactory vf = new ValueFactoryImpl(); + final MapBindingSet bindingSet = new MapBindingSet(); + bindingSet.addBinding("person", vf.createURI("urn:Alice")); + bindingSet.addBinding("age", vf.createLiteral(34)); + final VisibilityBindingSet visBs = new VisibilityBindingSet(bindingSet, "a"); + + // Mock the processor context that will be invoked. + final ProcessorContext context = mock(ProcessorContext.class); + + // Run the test. + final BindingSetOutputFormatter formatter = new BindingSetOutputFormatter(); + formatter.init(context); + formatter.process("key", ProcessorResult.make(new UnaryResult(visBs))); + + // Verify the mock was invoked with the expected output. + verify(context, times(1)).forward(eq("key"), eq(visBs)); + } + + @Test + public void binaryResult() { + // Create the input binding set. + final ValueFactory vf = new ValueFactoryImpl(); + final MapBindingSet bindingSet = new MapBindingSet(); + bindingSet.addBinding("person", vf.createURI("urn:Alice")); + bindingSet.addBinding("age", vf.createLiteral(34)); + final VisibilityBindingSet visBs = new VisibilityBindingSet(bindingSet, "a"); + + // Mock the processor context that will be invoked. + final ProcessorContext context = mock(ProcessorContext.class); + + // Run the test. + final BindingSetOutputFormatter formatter = new BindingSetOutputFormatter(); + formatter.init(context); + formatter.process("key", ProcessorResult.make(new BinaryResult(Side.LEFT, visBs))); + + // Verify the mock was invoked with the expected output. + verify(context, times(1)).forward(eq("key"), eq(visBs)); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/b8b0a126/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/output/StatementOutputFormatterTest.java ---------------------------------------------------------------------- diff --git a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/output/StatementOutputFormatterTest.java b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/output/StatementOutputFormatterTest.java new file mode 100644 index 0000000..5f7e9a6 --- /dev/null +++ b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/processors/output/StatementOutputFormatterTest.java @@ -0,0 +1,203 @@ +/* + * 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.streams.kafka.processors.output; + +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import org.apache.kafka.streams.processor.ProcessorContext; +import org.apache.rya.api.model.VisibilityBindingSet; +import org.apache.rya.api.model.VisibilityStatement; +import org.apache.rya.streams.kafka.processors.ProcessorResult; +import org.apache.rya.streams.kafka.processors.ProcessorResult.BinaryResult; +import org.apache.rya.streams.kafka.processors.ProcessorResult.BinaryResult.Side; +import org.apache.rya.streams.kafka.processors.ProcessorResult.UnaryResult; +import org.apache.rya.streams.kafka.processors.output.StatementOutputFormatterSupplier.StatementOutputFormatter; +import org.junit.Test; +import org.openrdf.model.ValueFactory; +import org.openrdf.model.impl.ValueFactoryImpl; +import org.openrdf.query.impl.MapBindingSet; + +/** + * Unit tests the methods of {@link StatementOutputFormatter}. + */ +public class StatementOutputFormatterTest { + + @Test + public void unaryResult() { + // Create the input binding set. + final ValueFactory vf = new ValueFactoryImpl(); + final MapBindingSet bindingSet = new MapBindingSet(); + bindingSet.addBinding("subject", vf.createURI("urn:Alice")); + bindingSet.addBinding("predicate", vf.createURI("urn:age")); + bindingSet.addBinding("object", vf.createLiteral(34)); + final VisibilityBindingSet visBs = new VisibilityBindingSet(bindingSet, "a"); + + // Mock the processor context that will be invoked. + final ProcessorContext context = mock(ProcessorContext.class); + + // Run the test. + final StatementOutputFormatter formatter = new StatementOutputFormatter(); + formatter.init(context); + formatter.process("key", ProcessorResult.make(new UnaryResult(visBs))); + + // Verify the mock was invoked with the expected output. + final VisibilityStatement expectedStmt = new VisibilityStatement(vf.createStatement( + vf.createURI("urn:Alice"), + vf.createURI("urn:age"), + vf.createLiteral(34)), "a"); + verify(context, times(1)).forward(eq("key"), eq(expectedStmt)); + } + + @Test + public void binaryResult() { + // Create the input binding set. + final ValueFactory vf = new ValueFactoryImpl(); + final MapBindingSet bindingSet = new MapBindingSet(); + bindingSet.addBinding("subject", vf.createURI("urn:Alice")); + bindingSet.addBinding("predicate", vf.createURI("urn:age")); + bindingSet.addBinding("object", vf.createLiteral(34)); + final VisibilityBindingSet visBs = new VisibilityBindingSet(bindingSet, "a"); + + // Mock the processor context that will be invoked. + final ProcessorContext context = mock(ProcessorContext.class); + + // Run the test. + final StatementOutputFormatter formatter = new StatementOutputFormatter(); + formatter.init(context); + formatter.process("key", ProcessorResult.make(new BinaryResult(Side.LEFT, visBs))); + + // Verify the mock was invoked with the expected output. + final VisibilityStatement expectedStmt = new VisibilityStatement(vf.createStatement( + vf.createURI("urn:Alice"), + vf.createURI("urn:age"), + vf.createLiteral(34)), "a"); + verify(context, times(1)).forward(eq("key"), eq(expectedStmt)); + } + + @Test + public void missingSubject() { + // Create the input binding set. + final ValueFactory vf = new ValueFactoryImpl(); + final MapBindingSet bindingSet = new MapBindingSet(); + bindingSet.addBinding("predicate", vf.createURI("urn:age")); + bindingSet.addBinding("object", vf.createLiteral(34)); + final VisibilityBindingSet visBs = new VisibilityBindingSet(bindingSet, "a"); + + // Mock the processor context that will be invoked. + final ProcessorContext context = mock(ProcessorContext.class); + + // Run the test. + final StatementOutputFormatter formatter = new StatementOutputFormatter(); + formatter.init(context); + formatter.process("key", ProcessorResult.make(new UnaryResult(visBs))); + + // Verify the mock was never invoked. + verify(context, times(0)).forward(any(), any()); + } + + @Test + public void subjectWrongType() { + // Create the input binding set. + final ValueFactory vf = new ValueFactoryImpl(); + final MapBindingSet bindingSet = new MapBindingSet(); + bindingSet.addBinding("subject", vf.createLiteral("Alice")); + bindingSet.addBinding("predicate", vf.createURI("urn:age")); + bindingSet.addBinding("object", vf.createLiteral(34)); + final VisibilityBindingSet visBs = new VisibilityBindingSet(bindingSet, "a"); + + // Mock the processor context that will be invoked. + final ProcessorContext context = mock(ProcessorContext.class); + + // Run the test. + final StatementOutputFormatter formatter = new StatementOutputFormatter(); + formatter.init(context); + formatter.process("key", ProcessorResult.make(new UnaryResult(visBs))); + + // Verify the mock was never invoked. + verify(context, times(0)).forward(any(), any()); + } + + @Test + public void missingPredicate() { + // Create the input binding set. + final ValueFactory vf = new ValueFactoryImpl(); + final MapBindingSet bindingSet = new MapBindingSet(); + bindingSet.addBinding("subject", vf.createURI("urn:Alice")); + bindingSet.addBinding("object", vf.createLiteral(34)); + final VisibilityBindingSet visBs = new VisibilityBindingSet(bindingSet, "a"); + + // Mock the processor context that will be invoked. + final ProcessorContext context = mock(ProcessorContext.class); + + // Run the test. + final StatementOutputFormatter formatter = new StatementOutputFormatter(); + formatter.init(context); + formatter.process("key", ProcessorResult.make(new UnaryResult(visBs))); + + // Verify the mock was never invoked. + verify(context, times(0)).forward(any(), any()); + } + + @Test + public void predicateWrongType() { + // Create the input binding set. + final ValueFactory vf = new ValueFactoryImpl(); + final MapBindingSet bindingSet = new MapBindingSet(); + bindingSet.addBinding("subject", vf.createURI("urn:Alice")); + bindingSet.addBinding("predicate", vf.createLiteral("age")); + bindingSet.addBinding("object", vf.createLiteral(34)); + final VisibilityBindingSet visBs = new VisibilityBindingSet(bindingSet, "a"); + + // Mock the processor context that will be invoked. + final ProcessorContext context = mock(ProcessorContext.class); + + // Run the test. + final StatementOutputFormatter formatter = new StatementOutputFormatter(); + formatter.init(context); + formatter.process("key", ProcessorResult.make(new UnaryResult(visBs))); + + // Verify the mock was never invoked. + verify(context, times(0)).forward(any(), any()); + } + + @Test + public void missingObject() { + // Create the input binding set. + final ValueFactory vf = new ValueFactoryImpl(); + final MapBindingSet bindingSet = new MapBindingSet(); + bindingSet.addBinding("subject", vf.createURI("urn:Alice")); + bindingSet.addBinding("predicate", vf.createURI("urn:age")); + final VisibilityBindingSet visBs = new VisibilityBindingSet(bindingSet, "a"); + + // Mock the processor context that will be invoked. + final ProcessorContext context = mock(ProcessorContext.class); + + // Run the test. + final StatementOutputFormatter formatter = new StatementOutputFormatter(); + formatter.init(context); + formatter.process("key", ProcessorResult.make(new UnaryResult(visBs))); + + // Verify the mock was never invoked. + verify(context, times(0)).forward(any(), any()); + } +} \ No newline at end of file
