Close partitionMetadataProvider after use in computePartitionGroupMetadata().
@@ -82,12 +82,13 @@ public interface StreamMetadataProvider extends Closeable {
// Use offset criteria from stream config
StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
for (int i = partitionGroupConsumptionStatuses.size(); i < partitionCount; i++) {
- StreamMetadataProvider partitionMetadataProvider =
- streamConsumerFactory.createPartitionMetadataProvider(clientId, i);
+ try (StreamMetadataProvider partitionMetadataProvider =
+ streamConsumerFactory.createPartitionMetadataProvider(clientId, i)) {
StreamPartitionMsgOffset streamPartitionMsgOffset =
partitionMetadataProvider.fetchStreamPartitionOffset(streamConfig.getOffsetCriteria(), timeoutMillis);
newPartitionGroupMetadataList.add(new PartitionGroupMetadata(i, streamPartitionMsgOffset));
}
+ }
return newPartitionGroupMetadataList;
}
}
