reiabreu commented on code in PR #3679:
URL: https://github.com/apache/storm/pull/3679#discussion_r1745517153
##########
external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java:
##########
@@ -163,54 +164,56 @@ && isFirstPollOffsetStrategyIgnoringCommittedOffsets()) {
/**
* Emit a new batch.
*/
- public Map<String, Object> emitPartitionBatchNew(TransactionAttempt tx,
TridentCollector collector,
- KafkaTridentSpoutTopicPartition currBatchPartition, Map<String,
Object> lastBatch) {
+ public Map<KafkaTridentSpoutTopicPartition, Map<String, Object>>
emitNewBatch(TransactionAttempt tx,
+ TridentCollector collector, Set<KafkaTridentSpoutTopicPartition>
partitions,
+ Map<KafkaTridentSpoutTopicPartition, Map<String, Object>>
lastBatchMetaMap) {
- LOG.debug("Processing batch: [transaction = {}], [currBatchPartition =
{}], [lastBatchMetadata = {}], [collector = {}]",
- tx, currBatchPartition, lastBatch, collector);
+ LOG.debug("Processing batch: [transaction = {}], [currBatchPartitions
= {}], [lastBatchMetadata = {}], [collector = {}]",
+ tx, partitions, lastBatchMetaMap, collector);
- final TopicPartition currBatchTp =
currBatchPartition.getTopicPartition();
-
- throwIfEmittingForUnassignedPartition(currBatchTp);
-
- KafkaTridentSpoutBatchMetadata lastBatchMeta = lastBatch == null ?
null : KafkaTridentSpoutBatchMetadata.fromMap(lastBatch);
- KafkaTridentSpoutBatchMetadata currentBatch = lastBatchMeta;
- Collection<TopicPartition> pausedTopicPartitions =
Collections.emptySet();
-
- try {
- // pause other topic-partitions to only poll from current
topic-partition
- pausedTopicPartitions = pauseTopicPartitions(currBatchTp);
+ Map<KafkaTridentSpoutTopicPartition, Map<String, Object>> ret = new
HashMap<>();
Review Comment:
can you give a more explicit name to `ret?`
--
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: dev-unsubscr...@storm.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
