You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rya.apache.org by ca...@apache.org on 2018/01/09 21:48:35 UTC
[18/50] [abbrv] incubator-rya git commit: RYA-377 Implemented a Kafka
version of GetQueryResultStream.
RYA-377 Implemented a Kafka version of GetQueryResultStream.
Project: http://git-wip-us.apache.org/repos/asf/incubator-rya/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-rya/commit/fc9775e2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-rya/tree/fc9775e2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-rya/diff/fc9775e2
Branch: refs/heads/master
Commit: fc9775e2639b8efa263c4f54f32f233da7e0b397
Parents: 6056528
Author: kchilton2 <ke...@gmail.com>
Authored: Tue Oct 31 18:24:36 2017 -0400
Committer: caleb <ca...@parsons.com>
Committed: Tue Jan 9 15:13:00 2018 -0500
----------------------------------------------------------------------
.../streams/api/entity/QueryResultStream.java | 24 ++-
.../apache/rya/streams/kafka/KafkaTopics.java | 21 +-
.../kafka/entity/KafkaQueryResultStream.java | 99 ++++++++++
.../interactor/KafkaGetQueryResultStream.java | 108 ++++++++++
.../interactor/KafkaGetQueryResultStreamIT.java | 195 +++++++++++++++++++
.../rya/test/kafka/KafkaTestInstanceRule.java | 14 ++
6 files changed, 455 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/fc9775e2/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/entity/QueryResultStream.java
----------------------------------------------------------------------
diff --git a/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/entity/QueryResultStream.java b/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/entity/QueryResultStream.java
index fdd62df..aa5dcfd 100644
--- a/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/entity/QueryResultStream.java
+++ b/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/entity/QueryResultStream.java
@@ -18,7 +18,8 @@
*/
package org.apache.rya.streams.api.entity;
-import java.util.Collection;
+import static java.util.Objects.requireNonNull;
+
import java.util.UUID;
import org.apache.rya.api.model.VisibilityBindingSet;
@@ -31,12 +32,26 @@ import edu.umd.cs.findbugs.annotations.NonNull;
* An infinite stream of {@link VisibilityBindingSet}s that are the results of a query within Rya Streams.
*/
@DefaultAnnotation(NonNull.class)
-public interface QueryResultStream extends AutoCloseable {
+public abstract class QueryResultStream implements AutoCloseable {
+
+ private final UUID queryId;
+
+ /**
+ * Constructs an instance of {@link QueryResultStream}.
+ *
+ * @param queryId - The query whose results this stream iterates over. (not null)
+ */
+ public QueryResultStream(final UUID queryId) {
+ this.queryId = requireNonNull(queryId);
+ }
+
/**
* @return Identifies which query in Rya Streams this result stream is over.
*/
- public UUID getQueryId();
+ public UUID getQueryId() {
+ return queryId;
+ }
/**
* Wait at most {@code timeoutMs} milliseconds for the next collection of results.
@@ -44,7 +59,8 @@ public interface QueryResultStream extends AutoCloseable {
* @param timeoutMs - The number of milliseconds to at most wait for the next collection of results. (not null)
* @return The next collection of {@link VisibilityBindingSet}s that are the result of the query. Empty if
* there where no new results within the timout period.
+ * @throws IllegalStateException If the stream has been closed.
* @throws RyaStreamsException Could not fetch the next set of results.
*/
- public Collection<VisibilityBindingSet> poll(long timeoutMs) throws RyaStreamsException;
+ public abstract Iterable<VisibilityBindingSet> poll(long timeoutMs) throws IllegalStateException, RyaStreamsException;
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/fc9775e2/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/KafkaTopics.java
----------------------------------------------------------------------
diff --git a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/KafkaTopics.java b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/KafkaTopics.java
index dfc4c9d..a8fbf23 100644
--- a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/KafkaTopics.java
+++ b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/KafkaTopics.java
@@ -18,6 +18,10 @@
*/
package org.apache.rya.streams.kafka;
+import static java.util.Objects.requireNonNull;
+
+import java.util.UUID;
+
import org.apache.rya.streams.api.queries.QueryChangeLog;
import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
@@ -30,23 +34,36 @@ import edu.umd.cs.findbugs.annotations.NonNull;
public class KafkaTopics {
/**
- * Creates the Kafka topic that will be used for a specific instance of Rya's {@link QueryChangeLog}.
+ * Creates the Kafka topic name that is used for a specific instance of Rya's {@link QueryChangeLog}.
*
* @param ryaInstance - The Rya instance the change log is for. (not null)
* @return The name of the Kafka topic.
*/
public static String queryChangeLogTopic(final String ryaInstance) {
+ requireNonNull(ryaInstance);
return ryaInstance + "-QueryChangeLog";
}
/**
- * Creates the Kafka topic that will be used to load statements into the Rya Streams system for a specific
+ * Creates the Kafka topic name that is used to load statements into the Rya Streams system for a specific
* instance of Rya.
*
* @param ryaInstance - The Rya instance the statements are for. (not null)
* @return The name of the Kafka topic.
*/
public static String statementsTopic(final String ryaInstance) {
+ requireNonNull(ryaInstance);
return ryaInstance + "-Statements";
}
+
+ /**
+ * Creates the Kafka topic name that is used for a specific query that is managed within Rya Streams.
+ *
+ * @param queryId - The id of the query the topic is for.
+ * @return The name of the Kafka topic.
+ */
+ public static String queryResultsTopic(final UUID queryId) {
+ requireNonNull(queryId);
+ return "QueryResults-" + queryId.toString();
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/fc9775e2/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/entity/KafkaQueryResultStream.java
----------------------------------------------------------------------
diff --git a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/entity/KafkaQueryResultStream.java b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/entity/KafkaQueryResultStream.java
new file mode 100644
index 0000000..360aaa2
--- /dev/null
+++ b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/entity/KafkaQueryResultStream.java
@@ -0,0 +1,99 @@
+/*
+ * 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.entity;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.Iterator;
+import java.util.UUID;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.rya.api.model.VisibilityBindingSet;
+import org.apache.rya.streams.api.entity.QueryResultStream;
+import org.apache.rya.streams.api.exception.RyaStreamsException;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * A Kafka implementation of {@link QueryResultStream}. It delegates the {@link #poll(long)} method to
+ * a {@link Consumer}. As a result, the starting point of this stream is whatever position the consumer
+ * starts at within the Kafka topic.
+ */
+@DefaultAnnotation(NonNull.class)
+public class KafkaQueryResultStream extends QueryResultStream {
+
+ private final Consumer<?, VisibilityBindingSet> consumer;
+
+ /**
+ * Constructs an instance of {@link KafkaQueryResultStream}.
+ *
+ * @param queryId - The query the results are for. (not null)
+ * @param consumer - The consumer that will be polled by this class. (not null)
+ */
+ public KafkaQueryResultStream(final UUID queryId, final Consumer<?, VisibilityBindingSet> consumer) {
+ super(queryId);
+ this.consumer = requireNonNull(consumer);
+ }
+
+ @Override
+ public Iterable<VisibilityBindingSet> poll(final long timeoutMs) throws RyaStreamsException {
+ return new RecordEntryIterable<>( consumer.poll(timeoutMs) );
+ }
+
+ /**
+ * An {@link Iterable} that creates {@link Iterator}s over a {@link ConsumerRecords}' values.
+ * This is useful for when you don't care about the key portion of a record.
+ *
+ * @param <K> - The type of the record's key.
+ * @param <T> - The type of the record's value.
+ */
+ private final class RecordEntryIterable<K, T> implements Iterable<T> {
+
+ private final ConsumerRecords<K, T> records;
+
+ public RecordEntryIterable(final ConsumerRecords<K, T> records) {
+ this.records = requireNonNull(records);
+ }
+
+ @Override
+ public Iterator<T> iterator() {
+ return new Iterator<T>() {
+ private final Iterator<ConsumerRecord<K, T>> it = records.iterator();
+
+ @Override
+ public boolean hasNext() {
+ return it.hasNext();
+ }
+
+ @Override
+ public T next() {
+ return it.next().value();
+ }
+ };
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ consumer.close();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/fc9775e2/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStream.java
----------------------------------------------------------------------
diff --git a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStream.java b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStream.java
new file mode 100644
index 0000000..b3c3fea
--- /dev/null
+++ b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStream.java
@@ -0,0 +1,108 @@
+/*
+ * 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.interactor;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.Arrays;
+import java.util.Properties;
+import java.util.UUID;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.rya.api.model.VisibilityBindingSet;
+import org.apache.rya.streams.api.entity.QueryResultStream;
+import org.apache.rya.streams.api.exception.RyaStreamsException;
+import org.apache.rya.streams.api.interactor.GetQueryResultStream;
+import org.apache.rya.streams.kafka.KafkaTopics;
+import org.apache.rya.streams.kafka.entity.KafkaQueryResultStream;
+import org.apache.rya.streams.kafka.serialization.VisibilityBindingSetDeserializer;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * A Kafka topic implementation of {@link GetQueryResultStream}.
+ */
+@DefaultAnnotation(NonNull.class)
+public class KafkaGetQueryResultStream implements GetQueryResultStream {
+
+ private final String bootstrapServers;
+
+ /**
+ * Constructs an instance of {@link KafkaGetQueryResultStream}.
+ *
+ * @param kafkaHostname - The hostname of the Kafka Broker to connect to. (not null)
+ * @param kafkaPort - The port of the Kafka Broker to connect to. (not null)
+ */
+ public KafkaGetQueryResultStream(final String kafkaHostname, final String kafkaPort) {
+ requireNonNull(kafkaHostname);
+ requireNonNull(kafkaPort);
+ bootstrapServers = kafkaHostname + ":" + kafkaPort;
+ }
+
+ @Override
+ public QueryResultStream fromStart(final UUID queryId) throws RyaStreamsException {
+ requireNonNull(queryId);
+
+ // Always start at the earliest point within the topic.
+ return makeStream(queryId, "earliest");
+ }
+
+ @Override
+ public QueryResultStream fromNow(final UUID queryId) throws RyaStreamsException {
+ requireNonNull(queryId);
+
+ // Always start at the latest point within the topic.
+ return makeStream(queryId, "latest");
+ }
+
+ private QueryResultStream makeStream(final UUID queryId, final String autoOffsetResetConfig) {
+ // Configure which instance of Kafka to connect to.
+ final Properties props = new Properties();
+ props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+
+ // Nothing meaningful is in the key and the value is a VisibilityBindingSet.
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, VisibilityBindingSetDeserializer.class);
+
+ // Use a UUID for the Group Id so that we never register as part of the same group as another consumer.
+ props.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
+
+ // Set a client id so that server side logging can be traced.
+ props.put(ConsumerConfig.CLIENT_ID_CONFIG, "Query-Result-Stream-" + queryId);
+
+ // These consumers always start at a specific point and move forwards until the caller is finished with
+ // the returned stream, so never commit the consumer's progress.
+ props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetResetConfig);
+ props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+
+ // We are not closing the consumer here because the returned QueryResultStream is responsible for closing the
+ // underlying resources required to process it.
+ final KafkaConsumer<Object, VisibilityBindingSet> consumer = new KafkaConsumer<>(props);
+
+ // Register the consumer for the query's results.
+ final String resultTopic = KafkaTopics.queryResultsTopic(queryId);
+ consumer.subscribe(Arrays.asList(resultTopic));
+
+ // Return the result stream.
+ return new KafkaQueryResultStream(queryId, consumer);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/fc9775e2/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStreamIT.java
----------------------------------------------------------------------
diff --git a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStreamIT.java b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStreamIT.java
new file mode 100644
index 0000000..3343f76
--- /dev/null
+++ b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStreamIT.java
@@ -0,0 +1,195 @@
+/*
+ * 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.interactor;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.UUID;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.rya.api.model.VisibilityBindingSet;
+import org.apache.rya.streams.api.entity.QueryResultStream;
+import org.apache.rya.streams.api.interactor.GetQueryResultStream;
+import org.apache.rya.streams.kafka.KafkaTopics;
+import org.apache.rya.streams.kafka.serialization.VisibilityBindingSetSerializer;
+import org.apache.rya.test.kafka.KafkaTestInstanceRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.openrdf.model.ValueFactory;
+import org.openrdf.model.impl.ValueFactoryImpl;
+import org.openrdf.query.impl.MapBindingSet;
+
+/**
+ * Integration tests the methods of {@link KafkaGetQueryResultStream}.
+ */
+public class KafkaGetQueryResultStreamIT {
+
+ @Rule
+ public KafkaTestInstanceRule kafka = new KafkaTestInstanceRule(true);
+
+ /**
+ * @return A {@link Producer} that is able to write {@link VisibilityBindingSet}s.
+ */
+ private Producer<?, VisibilityBindingSet> makeProducer() {
+ final Properties producerProps = kafka.createBootstrapServerConfig();
+ producerProps.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ producerProps.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, VisibilityBindingSetSerializer.class.getName());
+ return new KafkaProducer<>(producerProps);
+ }
+
+ /**
+ * Polls a {@link QueryResultStream} until it has either polled too many times without hitting
+ * the target number of results, or it hits the target number of results.
+ *
+ * @param pollMs - How long each poll could take.
+ * @param pollIterations - The maximum nubmer of polls that will be attempted.
+ * @param targetSize - The number of results to read before stopping.
+ * @param stream - The stream that will be polled.
+ * @return The results that were read from the stream.
+ * @throws Exception If the poll failed.
+ */
+ private List<VisibilityBindingSet> pollForResults(
+ final int pollMs,
+ final int pollIterations,
+ final int targetSize,
+ final QueryResultStream stream) throws Exception{
+ final List<VisibilityBindingSet> read = new ArrayList<>();
+
+ int i = 0;
+ while(read.size() < targetSize && i < pollIterations) {
+ for(final VisibilityBindingSet visBs : stream.poll(pollMs)) {
+ read.add( visBs );
+ }
+ i++;
+ }
+
+ return read;
+ }
+
+ @Test
+ public void fromStart() throws Exception {
+ // Create an ID for the query.
+ final UUID queryId = UUID.randomUUID();
+
+ // Create a list of test VisibilityBindingSets.
+ final List<VisibilityBindingSet> original = new ArrayList<>();
+
+ final ValueFactory vf = new ValueFactoryImpl();
+ MapBindingSet bs = new MapBindingSet();
+ bs.addBinding("urn:name", vf.createLiteral("Alice"));
+ original.add(new VisibilityBindingSet(bs, "a|b|c"));
+
+ bs = new MapBindingSet();
+ bs.addBinding("urn:name", vf.createLiteral("Bob"));
+ original.add(new VisibilityBindingSet(bs, "a"));
+
+ bs = new MapBindingSet();
+ bs.addBinding("urn:name", vf.createLiteral("Charlie"));
+ original.add(new VisibilityBindingSet(bs, "b|c"));
+
+ // Write some entries to the query result topic in Kafka.
+ try(final Producer<?, VisibilityBindingSet> producer = makeProducer()) {
+ final String resultTopic = KafkaTopics.queryResultsTopic(queryId);
+ for(final VisibilityBindingSet visBs : original) {
+ producer.send(new ProducerRecord<>(resultTopic, visBs));
+ }
+ }
+
+ // Use the interactor that is being tested to read all of the visibility binding sets.
+ final GetQueryResultStream interactor = new KafkaGetQueryResultStream(kafka.getKafkaHostname(), kafka.getKafkaPort());
+ final List<VisibilityBindingSet> read = pollForResults(500, 3, 3, interactor.fromStart(queryId));
+
+ // Show the fetched binding sets match the original, as well as their order.
+ assertEquals(original, read);
+ }
+
+ @Test
+ public void fromNow() throws Exception {
+ // Create an ID for the query.
+ final UUID queryId = UUID.randomUUID();
+
+ try(final Producer<?, VisibilityBindingSet> producer = makeProducer()) {
+ final String resultTopic = KafkaTopics.queryResultsTopic(queryId);
+
+ // Write a single visibility binding set to the query's result topic. This will not appear in the expected results.
+ final ValueFactory vf = new ValueFactoryImpl();
+ MapBindingSet bs = new MapBindingSet();
+ bs.addBinding("urn:name", vf.createLiteral("Alice"));
+ producer.send(new ProducerRecord<>(resultTopic, new VisibilityBindingSet(bs, "a|b|c")));
+ producer.flush();
+
+ // Use the interactor that is being tested to read all of the visibility binding sets that appear after this point.
+ final GetQueryResultStream interactor = new KafkaGetQueryResultStream(kafka.getKafkaHostname(), kafka.getKafkaPort());
+ try(QueryResultStream results = interactor.fromNow(queryId)) {
+ // Read results from the stream.
+ List<VisibilityBindingSet> read = new ArrayList<>();
+ for(final VisibilityBindingSet visBs : results.poll(500)) {
+ read.add(visBs);
+ }
+
+ // Show nothing has been read.
+ assertTrue(read.isEmpty());
+
+ // Write two more entries to the result topic. These will be seen by the result stream.
+ final List<VisibilityBindingSet> original = new ArrayList<>();
+
+ bs = new MapBindingSet();
+ bs.addBinding("urn:name", vf.createLiteral("Bob"));
+ original.add(new VisibilityBindingSet(bs, "a"));
+
+ bs = new MapBindingSet();
+ bs.addBinding("urn:name", vf.createLiteral("Charlie"));
+ original.add(new VisibilityBindingSet(bs, "b|c"));
+
+ for(final VisibilityBindingSet visBs : original) {
+ producer.send(new ProducerRecord<>(resultTopic, visBs));
+ }
+ producer.flush();
+
+ // Read the results from the result stream.
+ read = pollForResults(500, 3, 2, results);
+
+ // Show the new entries were read.
+ assertEquals(original, read);
+ }
+ }
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void pollClosedStream() throws Exception {
+ // Create an ID for the query.
+ final UUID queryId = UUID.randomUUID();
+
+ // Use the interactor that is being tested to create a result stream and immediately close it.
+ final GetQueryResultStream interactor = new KafkaGetQueryResultStream(kafka.getKafkaHostname(), kafka.getKafkaPort());
+ final QueryResultStream results = interactor.fromStart(queryId);
+ results.close();
+
+ // Try to poll the closed stream.
+ results.poll(1);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/fc9775e2/test/kafka/src/main/java/org/apache/rya/test/kafka/KafkaTestInstanceRule.java
----------------------------------------------------------------------
diff --git a/test/kafka/src/main/java/org/apache/rya/test/kafka/KafkaTestInstanceRule.java b/test/kafka/src/main/java/org/apache/rya/test/kafka/KafkaTestInstanceRule.java
index 41a0a67..f76fa2b 100644
--- a/test/kafka/src/main/java/org/apache/rya/test/kafka/KafkaTestInstanceRule.java
+++ b/test/kafka/src/main/java/org/apache/rya/test/kafka/KafkaTestInstanceRule.java
@@ -103,4 +103,18 @@ public class KafkaTestInstanceRule extends ExternalResource {
public Properties createBootstrapServerConfig() {
return kafkaInstance.createBootstrapServerConfig();
}
+
+ /**
+ * @return The hostname of the Kafka Broker.
+ */
+ public String getKafkaHostname() {
+ return kafkaInstance.getBrokerHost();
+ }
+
+ /**
+ * @return The port of the Kafka Broker.
+ */
+ public String getKafkaPort() {
+ return kafkaInstance.getBrokerPort();
+ }
}
\ No newline at end of file