kfaraz commented on a change in pull request #12235:
URL: https://github.com/apache/druid/pull/12235#discussion_r803306040
##########
File path:
extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisor.java
##########
@@ -416,6 +424,49 @@ protected boolean supportsPartitionExpiration()
return true;
}
+ @Override
+ protected boolean shouldSkipIgnorablePartitions()
+ {
+ return spec.getSpec().getTuningConfig().shouldSkipIgnorableShards();
+ }
+
+ /**
+ * Closed and empty shards can be ignored for ingestion,
+ * Use this method if skipIgnorablePartitions is true in the spec
+ *
+ * These partitions can be safely ignored for both ingestion task assignment
and autoscaler limits
+ *
+ * @return the set of ignorable shards' ids
+ */
+ @Override
+ protected Set<String> getIgnorablePartitionIds()
+ {
+ updateClosedShardCache();
+ return getEmptyClosedShardIds();
+ }
+
+ private void updateClosedShardCache()
+ {
+ String stream = spec.getSource();
+ Set<Shard> allActiveShards = ((KinesisRecordSupplier)
recordSupplier).getShards(stream);
+ Set<String> activeClosedShards = allActiveShards.stream()
+ .filter(shard ->
isShardClosed(shard))
+
.map(Shard::getShardId).collect(Collectors.toSet());
+
+ // clear stale shards
+ emptyClosedShardIds.retainAll(activeClosedShards);
Review comment:
Shouldn't we just clear these two sets here?
They are being re-populated in the for loop later anyway.
##########
File path:
extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisor.java
##########
@@ -481,4 +532,53 @@ protected boolean supportsPartitionExpiration()
return new KinesisDataSourceMetadata(newSequences);
}
+
+ /**
+ * Closed shards iff they have an ending sequence number
Review comment:
```suggestion
* A shard is considered closed iff it has an ending sequence number.
```
##########
File path:
extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTuningConfig.java
##########
@@ -212,6 +216,12 @@ public Duration getOffsetFetchPeriod()
return offsetFetchPeriod;
}
+ @JsonProperty
+ public boolean shouldSkipIgnorableShards()
Review comment:
```suggestion
public boolean isSkipIgnorableShards()
```
Since this is a serializable POJO, best to adhere to method naming
conventions.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
