dxichen commented on a change in pull request #1581:
URL: https://github.com/apache/samza/pull/1581#discussion_r798899937
##########
File path:
samza-test/src/test/java/org/apache/samza/test/framework/StreamApplicationIntegrationTestHarness.java
##########
@@ -128,32 +129,32 @@ public int clusterSize() {
return Integer.parseInt(KafkaConfig.TOPIC_DEFAULT_REPLICATION_FACTOR());
}
-
/**
* Read messages from the provided list of topics until {@param threshold}
messages have been read or until
* {@link #numEmptyPolls} polls return no messages.
*
* The default poll time out is determined by {@link #POLL_TIMEOUT_MS} and
the number of empty polls are
* determined by {@link #numEmptyPolls}
*
- * @param topics the list of topics to consume from
+ * @param topic the topic to consume from
* @param threshold the number of messages to consume
* @return the list of {@link ConsumerRecord}s whose size can be atmost
{@param threshold}
*/
- public List<ConsumerRecord<String, String>>
consumeMessages(Collection<String> topics, int threshold) {
+ public List<ConsumerRecord<String, String>> consumeMessages(String topic,
int threshold) {
int emptyPollCount = 0;
List<ConsumerRecord<String, String>> recordList = new ArrayList<>();
- consumer.subscribe(topics);
Review comment:
Sorry should have been more clear, I was talking about the `consumer`
here which is from `IntegrationTestHarness` which this class extends, which has
multiple uses except for this one.
Do we need that `consumer` to be `kafkaConsumer` map instead?
I wanted to verify that this class does not have any conflict with the base
class when using different consumers
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
