You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@streampipes.apache.org by "Yuukadesu (via GitHub)" <gi...@apache.org> on 2025/09/28 03:10:43 UTC
[PR] feat:Add SpRateLimiter and SpMemoryManager
Yuukadesu opened a new pull request, #3796:
URL: https://github.com/apache/streampipes/pull/3796
<!--
~ 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.
~
-->
<!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
- https://streampipes.apache.org/community/get-involved/
- https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar.
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
- http://apache.org/licenses/icla.pdf
-->
### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
-->
### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): <yes/no>
PR introduces (a) deprecation(s): <yes/no>
--
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-unsubscribe@streampipes.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat:Add SpRateLimiter and SpMemoryManager
Posted by "luoluoyuyu (via GitHub)" <gi...@apache.org>.
luoluoyuyu commented on code in PR #3796:
URL: https://github.com/apache/streampipes/pull/3796#discussion_r2455091844
##########
streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/limiter/SpRateLimiter.java:
##########
@@ -0,0 +1,340 @@
+/*
+ * 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.streampipes.extensions.api.limiter;
+
+import org.apache.streampipes.commons.prometheus.spratelimiter.SpRateLimiterStats;
+
+import com.google.common.util.concurrent.RateLimiter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectOutputStream;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * A singleton rate limiter implementation for StreamPipes extensions.
+ * This class provides rate limiting functionality using Google Guava's RateLimiter.
+ * It supports configurable permits per second and warmup periods.
+ */
+public enum SpRateLimiter {
+
+ INSTANCE;
+
+ private static final Logger LOG = LoggerFactory.getLogger(SpRateLimiter.class);
+
+ // Configuration constants
+ private static final double DEFAULT_PERMITS_PER_SECOND = 100.0;
+ private static final long DEFAULT_WARMUP_PERIOD = 1000L;
+ private static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS;
+ private static final int SCHEDULER_INITIAL_DELAY_SECONDS = 0;
+ private static final int SCHEDULER_PERIOD_SECONDS = 15;
+ private static final long ZERO_RATE_WAIT_TIME_MS = 1000L;
+ private static final int STATS_RESET_THRESHOLD = 1000;
+ private static final int STATS_RESET_FACTOR = 999;
+ private static final int STATS_RESET_DIVISOR = 1000;
+ private static final int SHUTDOWN_TIMEOUT_SECONDS = 5;
+ private static final long TIMEOUT_MS = 1000;
+ private static final int PERMITS_PER_REQUEST = 1;
+
+ private RateLimiter rateLimiter;
+
+ private double rateLimiterAverageWaitTime = 0.0;
+
+ private long totalWaitTime = 0L;
+ private AtomicInteger waitTimeCount = new AtomicInteger(0);
+
+ private AtomicInteger currentQueueSize = new AtomicInteger(0);
+
+ private SpRateLimiterStats stats;
+ private static volatile boolean schedulerInitialized = false;
+ private static ScheduledExecutorService scheduler;
+
+ /**
+ * Creates a rate limiter with default parameters.
+ * Default: 100 permits per second, 1000ms warmup period.
+ */
+ public void createRateLimiter() {
+ createRateLimiter(DEFAULT_PERMITS_PER_SECOND, DEFAULT_WARMUP_PERIOD, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created and scheduler initialized");
+ }
+
+ /**
+ * Creates a rate limiter with the specified permits per second.
+ * Uses default warmup period of 1000ms.
+ *
+ * @param permitsPerSecond The number of permits per second
+ */
+ public void createRateLimiter(double permitsPerSecond) {
+ createRateLimiter(permitsPerSecond, DEFAULT_WARMUP_PERIOD, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created with {} permits/sec and scheduler initialized", permitsPerSecond);
+ }
+
+ public void initScheduledTasks() {
+ if (!schedulerInitialized) {
+ synchronized (SpRateLimiter.class) {
+ if (!schedulerInitialized) {
+ scheduler = Executors.newSingleThreadScheduledExecutor();
+ scheduler.scheduleAtFixedRate(this::scheduledTask, SCHEDULER_INITIAL_DELAY_SECONDS, SCHEDULER_PERIOD_SECONDS, TimeUnit.SECONDS);
+ schedulerInitialized = true;
+ }
+ }
+ }
+ }
+
+ public void scheduledTask() {
+ this.stats = new SpRateLimiterStats();
+ stats.setAverageWaitTime(this.rateLimiterAverageWaitTime);
+ stats.setQueueSize(this.currentQueueSize.get());
+ stats.updateAllMetrics();
+ }
+
+ /**
+ * Creates a rate limiter with the specified parameters.
+ *
+ * @param permitsPerSecond The number of permits per second
+ * @param warmupPeriod The warmup period
+ * @param unit The time unit for the warmup period
+ * @throws IllegalArgumentException if parameters are invalid
+ */
+ public void createRateLimiter(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
+ if (this.rateLimiter == null) {
+ validateParameters(permitsPerSecond, warmupPeriod, unit);
+ this.rateLimiter = RateLimiter.create(permitsPerSecond, warmupPeriod, unit);
+ LOG.info("RateLimiter created with {} permits per second, warmup period: {} {}",
+ permitsPerSecond, warmupPeriod, unit);
+ } else {
+ LOG.warn("RateLimiter already exists. Use setRate() to modify the rate instead.");
+ }
+ }
+
+ /**
+ * Acquires a permit from the rate limiter for processing data, with timeout.
+ * Each request consumes exactly 1 permit regardless of data size.
+ * This provides simple and fair rate limiting based on request count.
+ *
+ * @param bytes The number of bytes to process (for logging purposes only)
+ * @return true if permit was acquired successfully, false if timeout occurred
+ * @throws InterruptedException if the current thread is interrupted while waiting
+ */
+ public boolean limit(long bytes) throws InterruptedException {
+ if (this.rateLimiter == null) {
+ LOG.warn("RateLimiter has not been initialized. Please call createRateLimiter() first.");
+ return false;
+ }
+
+ long startTime = System.currentTimeMillis();
+
+ synchronized (this) {
+ currentQueueSize.incrementAndGet();
+ }
+
+ try {
+ if (rateLimiter.getRate() <= 0) {
+ LOG.warn("RateLimiter is set to zero or negative rate. No permits will be acquired.");
+ updateAverageWaitTime(ZERO_RATE_WAIT_TIME_MS);
+ try {
+ Thread.sleep(ZERO_RATE_WAIT_TIME_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ return false;
+ } else {
+ // Each request consumes exactly 1 permit regardless of data size
+ long timeoutMs = TIMEOUT_MS;
+ boolean acquired = rateLimiter.tryAcquire(PERMITS_PER_REQUEST, timeoutMs, TimeUnit.MILLISECONDS);
+
+ long waitTime = System.currentTimeMillis() - startTime;
+ updateAverageWaitTime(waitTime);
+
+ if (!acquired) {
+ LOG.warn("Failed to acquire permit for {} bytes within {} ms timeout (rate: {} requests/sec)",
+ bytes, timeoutMs, rateLimiter.getRate());
+ } else {
+ LOG.debug("Successfully acquired permit for {} bytes in {} ms (rate: {} requests/sec)",
+ bytes, waitTime, rateLimiter.getRate());
+ }
+
+ return acquired;
+ }
+ } finally {
+ synchronized (this) {
Review Comment:
delete?
##########
streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/limiter/SpRateLimiter.java:
##########
@@ -0,0 +1,340 @@
+/*
+ * 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.streampipes.extensions.api.limiter;
+
+import org.apache.streampipes.commons.prometheus.spratelimiter.SpRateLimiterStats;
+
+import com.google.common.util.concurrent.RateLimiter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectOutputStream;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * A singleton rate limiter implementation for StreamPipes extensions.
+ * This class provides rate limiting functionality using Google Guava's RateLimiter.
+ * It supports configurable permits per second and warmup periods.
+ */
+public enum SpRateLimiter {
+
+ INSTANCE;
+
+ private static final Logger LOG = LoggerFactory.getLogger(SpRateLimiter.class);
+
+ // Configuration constants
+ private static final double DEFAULT_PERMITS_PER_SECOND = 100.0;
+ private static final long DEFAULT_WARMUP_PERIOD = 1000L;
+ private static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS;
+ private static final int SCHEDULER_INITIAL_DELAY_SECONDS = 0;
+ private static final int SCHEDULER_PERIOD_SECONDS = 15;
+ private static final long ZERO_RATE_WAIT_TIME_MS = 1000L;
+ private static final int STATS_RESET_THRESHOLD = 1000;
+ private static final int STATS_RESET_FACTOR = 999;
+ private static final int STATS_RESET_DIVISOR = 1000;
+ private static final int SHUTDOWN_TIMEOUT_SECONDS = 5;
+ private static final long TIMEOUT_MS = 1000;
+ private static final int PERMITS_PER_REQUEST = 1;
+
+ private RateLimiter rateLimiter;
+
+ private double rateLimiterAverageWaitTime = 0.0;
+
+ private long totalWaitTime = 0L;
+ private AtomicInteger waitTimeCount = new AtomicInteger(0);
+
+ private AtomicInteger currentQueueSize = new AtomicInteger(0);
+
+ private SpRateLimiterStats stats;
+ private static volatile boolean schedulerInitialized = false;
+ private static ScheduledExecutorService scheduler;
+
+ /**
+ * Creates a rate limiter with default parameters.
+ * Default: 100 permits per second, 1000ms warmup period.
+ */
+ public void createRateLimiter() {
+ createRateLimiter(DEFAULT_PERMITS_PER_SECOND, DEFAULT_WARMUP_PERIOD, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created and scheduler initialized");
+ }
+
+ /**
+ * Creates a rate limiter with the specified permits per second.
+ * Uses default warmup period of 1000ms.
+ *
+ * @param permitsPerSecond The number of permits per second
+ */
+ public void createRateLimiter(double permitsPerSecond) {
+ createRateLimiter(permitsPerSecond, DEFAULT_WARMUP_PERIOD, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created with {} permits/sec and scheduler initialized", permitsPerSecond);
+ }
+
+ public void initScheduledTasks() {
+ if (!schedulerInitialized) {
+ synchronized (SpRateLimiter.class) {
+ if (!schedulerInitialized) {
+ scheduler = Executors.newSingleThreadScheduledExecutor();
+ scheduler.scheduleAtFixedRate(this::scheduledTask, SCHEDULER_INITIAL_DELAY_SECONDS, SCHEDULER_PERIOD_SECONDS, TimeUnit.SECONDS);
+ schedulerInitialized = true;
+ }
+ }
+ }
+ }
+
+ public void scheduledTask() {
+ this.stats = new SpRateLimiterStats();
+ stats.setAverageWaitTime(this.rateLimiterAverageWaitTime);
+ stats.setQueueSize(this.currentQueueSize.get());
+ stats.updateAllMetrics();
+ }
+
+ /**
+ * Creates a rate limiter with the specified parameters.
+ *
+ * @param permitsPerSecond The number of permits per second
+ * @param warmupPeriod The warmup period
+ * @param unit The time unit for the warmup period
+ * @throws IllegalArgumentException if parameters are invalid
+ */
+ public void createRateLimiter(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
+ if (this.rateLimiter == null) {
+ validateParameters(permitsPerSecond, warmupPeriod, unit);
+ this.rateLimiter = RateLimiter.create(permitsPerSecond, warmupPeriod, unit);
+ LOG.info("RateLimiter created with {} permits per second, warmup period: {} {}",
+ permitsPerSecond, warmupPeriod, unit);
+ } else {
+ LOG.warn("RateLimiter already exists. Use setRate() to modify the rate instead.");
+ }
+ }
+
+ /**
+ * Acquires a permit from the rate limiter for processing data, with timeout.
+ * Each request consumes exactly 1 permit regardless of data size.
+ * This provides simple and fair rate limiting based on request count.
+ *
+ * @param bytes The number of bytes to process (for logging purposes only)
+ * @return true if permit was acquired successfully, false if timeout occurred
+ * @throws InterruptedException if the current thread is interrupted while waiting
+ */
+ public boolean limit(long bytes) throws InterruptedException {
+ if (this.rateLimiter == null) {
+ LOG.warn("RateLimiter has not been initialized. Please call createRateLimiter() first.");
+ return false;
+ }
+
+ long startTime = System.currentTimeMillis();
+
+ synchronized (this) {
Review Comment:
delete?
##########
streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/limiter/SpRateLimiter.java:
##########
@@ -0,0 +1,340 @@
+/*
+ * 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.streampipes.extensions.api.limiter;
+
+import org.apache.streampipes.commons.prometheus.spratelimiter.SpRateLimiterStats;
+
+import com.google.common.util.concurrent.RateLimiter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectOutputStream;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * A singleton rate limiter implementation for StreamPipes extensions.
+ * This class provides rate limiting functionality using Google Guava's RateLimiter.
+ * It supports configurable permits per second and warmup periods.
+ */
+public enum SpRateLimiter {
+
+ INSTANCE;
+
+ private static final Logger LOG = LoggerFactory.getLogger(SpRateLimiter.class);
+
+ // Configuration constants
+ private static final double DEFAULT_PERMITS_PER_SECOND = 100.0;
+ private static final long DEFAULT_WARMUP_PERIOD = 1000L;
+ private static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS;
+ private static final int SCHEDULER_INITIAL_DELAY_SECONDS = 0;
+ private static final int SCHEDULER_PERIOD_SECONDS = 15;
+ private static final long ZERO_RATE_WAIT_TIME_MS = 1000L;
+ private static final int STATS_RESET_THRESHOLD = 1000;
+ private static final int STATS_RESET_FACTOR = 999;
+ private static final int STATS_RESET_DIVISOR = 1000;
+ private static final int SHUTDOWN_TIMEOUT_SECONDS = 5;
+ private static final long TIMEOUT_MS = 1000;
+ private static final int PERMITS_PER_REQUEST = 1;
+
+ private RateLimiter rateLimiter;
+
+ private double rateLimiterAverageWaitTime = 0.0;
+
+ private long totalWaitTime = 0L;
+ private AtomicInteger waitTimeCount = new AtomicInteger(0);
+
+ private AtomicInteger currentQueueSize = new AtomicInteger(0);
+
+ private SpRateLimiterStats stats;
+ private static volatile boolean schedulerInitialized = false;
+ private static ScheduledExecutorService scheduler;
+
+ /**
+ * Creates a rate limiter with default parameters.
+ * Default: 100 permits per second, 1000ms warmup period.
+ */
+ public void createRateLimiter() {
+ createRateLimiter(DEFAULT_PERMITS_PER_SECOND, DEFAULT_WARMUP_PERIOD, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created and scheduler initialized");
+ }
+
+ /**
+ * Creates a rate limiter with the specified permits per second.
+ * Uses default warmup period of 1000ms.
+ *
+ * @param permitsPerSecond The number of permits per second
+ */
+ public void createRateLimiter(double permitsPerSecond) {
+ createRateLimiter(permitsPerSecond, DEFAULT_WARMUP_PERIOD, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created with {} permits/sec and scheduler initialized", permitsPerSecond);
+ }
+
+ public void initScheduledTasks() {
+ if (!schedulerInitialized) {
+ synchronized (SpRateLimiter.class) {
+ if (!schedulerInitialized) {
+ scheduler = Executors.newSingleThreadScheduledExecutor();
+ scheduler.scheduleAtFixedRate(this::scheduledTask, SCHEDULER_INITIAL_DELAY_SECONDS, SCHEDULER_PERIOD_SECONDS, TimeUnit.SECONDS);
+ schedulerInitialized = true;
+ }
+ }
+ }
+ }
+
+ public void scheduledTask() {
+ this.stats = new SpRateLimiterStats();
+ stats.setAverageWaitTime(this.rateLimiterAverageWaitTime);
+ stats.setQueueSize(this.currentQueueSize.get());
+ stats.updateAllMetrics();
+ }
+
+ /**
+ * Creates a rate limiter with the specified parameters.
+ *
+ * @param permitsPerSecond The number of permits per second
+ * @param warmupPeriod The warmup period
+ * @param unit The time unit for the warmup period
+ * @throws IllegalArgumentException if parameters are invalid
+ */
+ public void createRateLimiter(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
+ if (this.rateLimiter == null) {
+ validateParameters(permitsPerSecond, warmupPeriod, unit);
+ this.rateLimiter = RateLimiter.create(permitsPerSecond, warmupPeriod, unit);
+ LOG.info("RateLimiter created with {} permits per second, warmup period: {} {}",
+ permitsPerSecond, warmupPeriod, unit);
+ } else {
+ LOG.warn("RateLimiter already exists. Use setRate() to modify the rate instead.");
+ }
+ }
+
+ /**
+ * Acquires a permit from the rate limiter for processing data, with timeout.
+ * Each request consumes exactly 1 permit regardless of data size.
+ * This provides simple and fair rate limiting based on request count.
+ *
+ * @param bytes The number of bytes to process (for logging purposes only)
+ * @return true if permit was acquired successfully, false if timeout occurred
+ * @throws InterruptedException if the current thread is interrupted while waiting
+ */
+ public boolean limit(long bytes) throws InterruptedException {
+ if (this.rateLimiter == null) {
+ LOG.warn("RateLimiter has not been initialized. Please call createRateLimiter() first.");
+ return false;
+ }
+
+ long startTime = System.currentTimeMillis();
+
+ synchronized (this) {
+ currentQueueSize.incrementAndGet();
+ }
+
+ try {
+ if (rateLimiter.getRate() <= 0) {
+ LOG.warn("RateLimiter is set to zero or negative rate. No permits will be acquired.");
+ updateAverageWaitTime(ZERO_RATE_WAIT_TIME_MS);
+ try {
+ Thread.sleep(ZERO_RATE_WAIT_TIME_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ return false;
Review Comment:
delete
##########
streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/memorymanager/SpMemoryManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.streampipes.extensions.api.memorymanager;
+
+import org.apache.streampipes.commons.prometheus.spmemorymanager.SpMemoryManagerStats;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectOutputStream;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+/**
+ * A singleton memory manager implementation for StreamPipes extensions.
+ * This class provides memory allocation and deallocation functionality with
+ * blocking behavior when insufficient memory is available.
+ */
+public enum SpMemoryManager {
+
+ INSTANCE;
+
+ private static final Logger LOG = LoggerFactory.getLogger(SpMemoryManager.class);
+
+ // Configuration constants
+ private static final long DEFAULT_INITIAL_MEMORY = 10L * 1024 * 1024 * 1024; // 10 GB
+ private static final long WAIT_TIMEOUT_MS = 1000L;
+ private static final int SCHEDULER_INITIAL_DELAY_SECONDS = 0;
+ private static final int SCHEDULER_PERIOD_SECONDS = 15;
+ private static final int BYTES_TO_MB = 1024 * 1024;
+ private static final int SHUTDOWN_TIMEOUT_SECONDS = 5;
+
+ // Memory control thresholds
+ private static final double MEMORY_USAGE_THRESHOLD = 0.9; // 90% threshold for blocking
+ private static final double MEMORY_WARNING_THRESHOLD = 0.8; // 80% warning threshold
+
+ private long freeMemory;
+
+ private double memoryUsedBytes = 0.0;
+ private double memoryAllocationRate = 0.0;
+
+ private long lastAllocationTime = System.currentTimeMillis();
+ private long totalAllocatedBytes = 0L;
+
+ // Memory control state
+ private volatile boolean memoryBlocked = false;
+ private volatile boolean memoryWarningActive = false;
+
+ private SpMemoryManagerStats stats = new SpMemoryManagerStats();
+ private static volatile boolean schedulerInitialized = false;
+ private static ScheduledExecutorService scheduler;
+
+ SpMemoryManager() {
+ this.freeMemory = DEFAULT_INITIAL_MEMORY;
+ initScheduledTask();
+ }
+
+ public void initScheduledTask() {
+ if (!schedulerInitialized) {
+ synchronized (SpMemoryManager.class) {
+ if (!schedulerInitialized) {
+ scheduler = Executors.newScheduledThreadPool(1);
+ scheduler.scheduleAtFixedRate(this::scheduledTask, SCHEDULER_INITIAL_DELAY_SECONDS, SCHEDULER_PERIOD_SECONDS, java.util.concurrent.TimeUnit.SECONDS);
+ schedulerInitialized = true;
+ }
+ }
+ }
+ }
+
+ public void scheduledTask() {
+ this.stats = new SpMemoryManagerStats();
+ stats.setAllocationRate(this.getMEMORY_ALLOCATION_RATE());
+ stats.setMemoryUsedBytes(this.getMEMORY_USED_BYTES());
+ stats.updateAllMetrics();
+
+ // Check memory usage thresholds
+ checkMemoryThresholds();
+ }
+
+ /**
+ * Checks memory usage against configured thresholds and updates blocking state.
+ */
+ private void checkMemoryThresholds() {
+ double memoryUsageRatio = (double) getAllocatedMemory() / DEFAULT_INITIAL_MEMORY;
+
+ if (memoryUsageRatio >= MEMORY_USAGE_THRESHOLD) {
+ if (!memoryBlocked) {
+ memoryBlocked = true;
+ LOG.warn("Memory usage reached {}% threshold. Blocking data consumption.",
+ (int)(MEMORY_USAGE_THRESHOLD * 100));
+ }
+ } else if (memoryUsageRatio <= MEMORY_WARNING_THRESHOLD) {
+ if (memoryBlocked) {
+ memoryBlocked = false;
+ LOG.info("Memory usage dropped below {}% threshold. Resuming data consumption.",
+ (int)(MEMORY_WARNING_THRESHOLD * 100));
+ }
+ }
+
+ // Update warning state
+ memoryWarningActive = memoryUsageRatio >= MEMORY_WARNING_THRESHOLD
+ && memoryUsageRatio < MEMORY_USAGE_THRESHOLD;
+ }
+
+ /**
+ * Allocates the specified amount of memory.
+ * If insufficient memory is available, this method will block until memory becomes available.
+ * If memory usage is above threshold, allocation will be blocked.
+ *
+ * @param bytes The number of bytes to allocate
+ * @throws IllegalArgumentException if bytes is non-positive
+ */
+ public void allocate(long bytes) {
+ if (bytes <= 0) {
+ LOG.warn("Attempted to allocate non-positive memory: {} bytes", bytes);
+ return;
+ }
+
+ // Check if memory is blocked due to threshold
+ if (memoryBlocked) {
+ LOG.warn("Memory allocation blocked due to high memory usage threshold. "
+ + "Current usage: {}%", getMemoryUsagePercentage());
+ return;
+ }
+
+ // Loop until enough memory is available
+ while (true) {
+ long currentFree = freeMemory;
+ long newFreeMemory = currentFree - bytes;
+
+ if (newFreeMemory >= 0) {
+ // Sufficient memory available, perform allocation
+ freeMemory = newFreeMemory;
+
+ long allocatedMemory = DEFAULT_INITIAL_MEMORY - freeMemory;
+ memoryUsedBytes = (double) allocatedMemory;
+
+ updateAllocationRate(bytes);
+
+ return;
+ } else {
+ // Insufficient memory, block and wait
+ LOG.warn("Not enough free memory to allocate {} bytes. Current free memory: {} bytes. "
+ + "Blocking allocation.", bytes, currentFree);
+
+ synchronized (this) {
+ try {
+ // Block and wait for memory to be freed
+ wait(WAIT_TIMEOUT_MS);
Review Comment:
this.wait?
##########
streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/memorymanager/SpMemoryManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.streampipes.extensions.api.memorymanager;
+
+import org.apache.streampipes.commons.prometheus.spmemorymanager.SpMemoryManagerStats;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectOutputStream;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+/**
+ * A singleton memory manager implementation for StreamPipes extensions.
+ * This class provides memory allocation and deallocation functionality with
+ * blocking behavior when insufficient memory is available.
+ */
+public enum SpMemoryManager {
+
+ INSTANCE;
+
+ private static final Logger LOG = LoggerFactory.getLogger(SpMemoryManager.class);
+
+ // Configuration constants
+ private static final long DEFAULT_INITIAL_MEMORY = 10L * 1024 * 1024 * 1024; // 10 GB
+ private static final long WAIT_TIMEOUT_MS = 1000L;
+ private static final int SCHEDULER_INITIAL_DELAY_SECONDS = 0;
+ private static final int SCHEDULER_PERIOD_SECONDS = 15;
+ private static final int BYTES_TO_MB = 1024 * 1024;
+ private static final int SHUTDOWN_TIMEOUT_SECONDS = 5;
+
+ // Memory control thresholds
+ private static final double MEMORY_USAGE_THRESHOLD = 0.9; // 90% threshold for blocking
+ private static final double MEMORY_WARNING_THRESHOLD = 0.8; // 80% warning threshold
+
+ private long freeMemory;
+
+ private double memoryUsedBytes = 0.0;
+ private double memoryAllocationRate = 0.0;
+
+ private long lastAllocationTime = System.currentTimeMillis();
+ private long totalAllocatedBytes = 0L;
+
+ // Memory control state
+ private volatile boolean memoryBlocked = false;
+ private volatile boolean memoryWarningActive = false;
+
+ private SpMemoryManagerStats stats = new SpMemoryManagerStats();
+ private static volatile boolean schedulerInitialized = false;
+ private static ScheduledExecutorService scheduler;
+
+ SpMemoryManager() {
+ this.freeMemory = DEFAULT_INITIAL_MEMORY;
+ initScheduledTask();
+ }
+
+ public void initScheduledTask() {
+ if (!schedulerInitialized) {
+ synchronized (SpMemoryManager.class) {
+ if (!schedulerInitialized) {
+ scheduler = Executors.newScheduledThreadPool(1);
+ scheduler.scheduleAtFixedRate(this::scheduledTask, SCHEDULER_INITIAL_DELAY_SECONDS, SCHEDULER_PERIOD_SECONDS, java.util.concurrent.TimeUnit.SECONDS);
+ schedulerInitialized = true;
+ }
+ }
+ }
+ }
+
+ public void scheduledTask() {
+ this.stats = new SpMemoryManagerStats();
+ stats.setAllocationRate(this.getMEMORY_ALLOCATION_RATE());
+ stats.setMemoryUsedBytes(this.getMEMORY_USED_BYTES());
+ stats.updateAllMetrics();
+
+ // Check memory usage thresholds
+ checkMemoryThresholds();
+ }
+
+ /**
+ * Checks memory usage against configured thresholds and updates blocking state.
+ */
+ private void checkMemoryThresholds() {
+ double memoryUsageRatio = (double) getAllocatedMemory() / DEFAULT_INITIAL_MEMORY;
+
+ if (memoryUsageRatio >= MEMORY_USAGE_THRESHOLD) {
+ if (!memoryBlocked) {
+ memoryBlocked = true;
+ LOG.warn("Memory usage reached {}% threshold. Blocking data consumption.",
+ (int)(MEMORY_USAGE_THRESHOLD * 100));
+ }
+ } else if (memoryUsageRatio <= MEMORY_WARNING_THRESHOLD) {
+ if (memoryBlocked) {
+ memoryBlocked = false;
+ LOG.info("Memory usage dropped below {}% threshold. Resuming data consumption.",
+ (int)(MEMORY_WARNING_THRESHOLD * 100));
+ }
+ }
+
+ // Update warning state
+ memoryWarningActive = memoryUsageRatio >= MEMORY_WARNING_THRESHOLD
+ && memoryUsageRatio < MEMORY_USAGE_THRESHOLD;
+ }
+
+ /**
+ * Allocates the specified amount of memory.
+ * If insufficient memory is available, this method will block until memory becomes available.
+ * If memory usage is above threshold, allocation will be blocked.
+ *
+ * @param bytes The number of bytes to allocate
+ * @throws IllegalArgumentException if bytes is non-positive
+ */
+ public void allocate(long bytes) {
+ if (bytes <= 0) {
+ LOG.warn("Attempted to allocate non-positive memory: {} bytes", bytes);
+ return;
+ }
+
+ // Check if memory is blocked due to threshold
+ if (memoryBlocked) {
+ LOG.warn("Memory allocation blocked due to high memory usage threshold. "
+ + "Current usage: {}%", getMemoryUsagePercentage());
+ return;
+ }
+
+ // Loop until enough memory is available
+ while (true) {
+ long currentFree = freeMemory;
+ long newFreeMemory = currentFree - bytes;
+
+ if (newFreeMemory >= 0) {
+ // Sufficient memory available, perform allocation
+ freeMemory = newFreeMemory;
+
+ long allocatedMemory = DEFAULT_INITIAL_MEMORY - freeMemory;
+ memoryUsedBytes = (double) allocatedMemory;
+
+ updateAllocationRate(bytes);
+
+ return;
+ } else {
+ // Insufficient memory, block and wait
+ LOG.warn("Not enough free memory to allocate {} bytes. Current free memory: {} bytes. "
+ + "Blocking allocation.", bytes, currentFree);
+
+ synchronized (this) {
+ try {
+ // Block and wait for memory to be freed
+ wait(WAIT_TIMEOUT_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.warn("Memory allocation blocking was interrupted");
+ return;
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Frees the specified amount of memory.
+ * This will notify any threads waiting for memory allocation.
+ *
+ * @param bytes The number of bytes to free
+ * @throws IllegalArgumentException if bytes is non-positive
+ */
+ public void free(long bytes) {
+ if (bytes <= 0) {
+ LOG.warn("Attempted to free non-positive memory: {} bytes", bytes);
+ return;
+ }
+
+ synchronized (this) {
+ freeMemory += bytes;
Review Comment:
use AtomicInt
##########
streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/limiter/SpRateLimiter.java:
##########
@@ -0,0 +1,340 @@
+/*
+ * 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.streampipes.extensions.api.limiter;
+
+import org.apache.streampipes.commons.prometheus.spratelimiter.SpRateLimiterStats;
+
+import com.google.common.util.concurrent.RateLimiter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectOutputStream;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * A singleton rate limiter implementation for StreamPipes extensions.
+ * This class provides rate limiting functionality using Google Guava's RateLimiter.
+ * It supports configurable permits per second and warmup periods.
+ */
+public enum SpRateLimiter {
+
+ INSTANCE;
+
+ private static final Logger LOG = LoggerFactory.getLogger(SpRateLimiter.class);
+
+ // Configuration constants
+ private static final double DEFAULT_PERMITS_PER_SECOND = 100.0;
+ private static final long DEFAULT_WARMUP_PERIOD = 1000L;
+ private static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS;
+ private static final int SCHEDULER_INITIAL_DELAY_SECONDS = 0;
+ private static final int SCHEDULER_PERIOD_SECONDS = 15;
+ private static final long ZERO_RATE_WAIT_TIME_MS = 1000L;
+ private static final int STATS_RESET_THRESHOLD = 1000;
+ private static final int STATS_RESET_FACTOR = 999;
+ private static final int STATS_RESET_DIVISOR = 1000;
+ private static final int SHUTDOWN_TIMEOUT_SECONDS = 5;
+ private static final long TIMEOUT_MS = 1000;
+ private static final int PERMITS_PER_REQUEST = 1;
+
+ private RateLimiter rateLimiter;
+
+ private double rateLimiterAverageWaitTime = 0.0;
+
+ private long totalWaitTime = 0L;
+ private AtomicInteger waitTimeCount = new AtomicInteger(0);
+
+ private AtomicInteger currentQueueSize = new AtomicInteger(0);
+
+ private SpRateLimiterStats stats;
+ private static volatile boolean schedulerInitialized = false;
+ private static ScheduledExecutorService scheduler;
+
+ /**
+ * Creates a rate limiter with default parameters.
+ * Default: 100 permits per second, 1000ms warmup period.
+ */
+ public void createRateLimiter() {
+ createRateLimiter(DEFAULT_PERMITS_PER_SECOND, DEFAULT_WARMUP_PERIOD, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created and scheduler initialized");
+ }
+
+ /**
+ * Creates a rate limiter with the specified permits per second.
+ * Uses default warmup period of 1000ms.
+ *
+ * @param permitsPerSecond The number of permits per second
+ */
+ public void createRateLimiter(double permitsPerSecond) {
+ createRateLimiter(permitsPerSecond, DEFAULT_WARMUP_PERIOD, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created with {} permits/sec and scheduler initialized", permitsPerSecond);
+ }
+
+ public void initScheduledTasks() {
+ if (!schedulerInitialized) {
+ synchronized (SpRateLimiter.class) {
+ if (!schedulerInitialized) {
+ scheduler = Executors.newSingleThreadScheduledExecutor();
+ scheduler.scheduleAtFixedRate(this::scheduledTask, SCHEDULER_INITIAL_DELAY_SECONDS, SCHEDULER_PERIOD_SECONDS, TimeUnit.SECONDS);
+ schedulerInitialized = true;
+ }
+ }
+ }
+ }
+
+ public void scheduledTask() {
+ this.stats = new SpRateLimiterStats();
+ stats.setAverageWaitTime(this.rateLimiterAverageWaitTime);
+ stats.setQueueSize(this.currentQueueSize.get());
+ stats.updateAllMetrics();
+ }
+
+ /**
+ * Creates a rate limiter with the specified parameters.
+ *
+ * @param permitsPerSecond The number of permits per second
+ * @param warmupPeriod The warmup period
+ * @param unit The time unit for the warmup period
+ * @throws IllegalArgumentException if parameters are invalid
+ */
+ public void createRateLimiter(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
+ if (this.rateLimiter == null) {
+ validateParameters(permitsPerSecond, warmupPeriod, unit);
+ this.rateLimiter = RateLimiter.create(permitsPerSecond, warmupPeriod, unit);
+ LOG.info("RateLimiter created with {} permits per second, warmup period: {} {}",
Review Comment:
rateLimiter.setRate(totalmem*70%)
##########
streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/function/StreamPipesFunction.java:
##########
@@ -117,6 +122,10 @@ public void process(Map<String, Object> rawEvent, String topicName) {
increaseCounter(sourceInfo.getSourceId());
} catch (RuntimeException e) {
addError(e);
+ } catch (InterruptedException e) {
+ throw new SpRuntimeException(e);
Review Comment:
SetInterruoted
--
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-unsubscribe@streampipes.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat:Add SpRateLimiter and SpMemoryManager
Posted by "Yuukadesu (via GitHub)" <gi...@apache.org>.
Yuukadesu commented on PR #3796:
URL: https://github.com/apache/streampipes/pull/3796#issuecomment-3420495147
I have added some metrics to monitor the SpRateLimiter and SpMemoryManager, and now these metrics can be observed through Grafana.
SpRateLimiter
Average wait time
<img width="986" height="307" alt="截屏2025-10-20 12 20 50" src="https://github.com/user-attachments/assets/ee8a3c64-3d07-411c-b571-8bace43e405e" />
Current size of the waiting queue
<img width="985" height="302" alt="截屏2025-10-20 12 21 15" src="https://github.com/user-attachments/assets/fb8585ba-5694-4b05-9763-8e66c8c44ce1" />
SpMemoryManager
Memory used bytes
<img width="990" height="306" alt="截屏2025-10-20 12 22 06" src="https://github.com/user-attachments/assets/13ec88bf-5f38-4d15-aeef-005adaa71703" />
Memory allocation rate
<img width="988" height="310" alt="截屏2025-10-20 12 22 42" src="https://github.com/user-attachments/assets/746d692b-3dea-440e-a6f8-79d4162c9b29" />
--
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-unsubscribe@streampipes.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat:Add SpRateLimiter and SpMemoryManager
Posted by "luoluoyuyu (via GitHub)" <gi...@apache.org>.
luoluoyuyu commented on code in PR #3796:
URL: https://github.com/apache/streampipes/pull/3796#discussion_r2462791901
##########
streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/limiter/SpRateLimiter.java:
##########
@@ -0,0 +1,333 @@
+/*
+ * 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.streampipes.extensions.api.limiter;
+
+import com.sun.management.OperatingSystemMXBean;
+import org.apache.streampipes.commons.environment.Environment;
+import org.apache.streampipes.commons.environment.Environments;
+import org.apache.streampipes.commons.prometheus.spratelimiter.SpRateLimiterStats;
+
+import com.google.common.util.concurrent.RateLimiter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectOutputStream;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * A singleton rate limiter implementation for StreamPipes extensions.
+ * This class provides rate limiting functionality using Google Guava's RateLimiter.
+ * It supports configurable permits per second and warmup periods.
+ */
+public enum SpRateLimiter {
+
+ INSTANCE;
+
+ private static final Logger LOG = LoggerFactory.getLogger(SpRateLimiter.class);
+
+ // Configuration constants
+ private static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS;
+ private final Environment env = Environments.getEnvironment();
+
+ private RateLimiter rateLimiter;
+
+ private double rateLimiterAverageWaitTime = 0.0;
+
+ private long totalWaitTime = 0L;
+ private final AtomicInteger waitTimeCount = new AtomicInteger(0);
+
+ private final AtomicInteger currentQueueSize = new AtomicInteger(0);
+
+ private SpRateLimiterStats stats;
+ private static volatile boolean schedulerInitialized = false;
+ private static ScheduledExecutorService scheduler;
+
+ /**
+ * Creates a rate limiter with default parameters.
+ * Default: calculated permits per second based on memory, 1000ms warmup period.
+ */
+ public void createRateLimiter() {
+ var defaultPermits = setPermit();
+ var defaultWarmupPeriod = env.getRateLimiterDefaultWarmupPeriod().getValueOrDefault();
+ createRateLimiter(defaultPermits, defaultWarmupPeriod, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created with {} permits/sec and scheduler initialized", defaultPermits);
+ }
+
+ /**
+ * Creates a rate limiter with the specified permits per second.
+ * Uses default warmup period of 1000ms.
+ *
+ * @param permitsPerSecond The number of permits per second
+ */
+ public void createRateLimiter(double permitsPerSecond) {
+ var defaultWarmupPeriod = env.getRateLimiterDefaultWarmupPeriod().getValueOrDefault();
+ createRateLimiter(permitsPerSecond, defaultWarmupPeriod, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created with {} permits/sec and scheduler initialized", permitsPerSecond);
+ }
+
+ public void initScheduledTasks() {
+ var schedulerInitialDelay = env.getRateLimiterSchedulerInitialDelaySeconds().getValueOrDefault();
+ var schedulerPeriod = env.getRateLimiterSchedulerPeriodSeconds().getValueOrDefault();
+ if (!schedulerInitialized) {
+ synchronized (SpRateLimiter.class) {
+ if (!schedulerInitialized) {
+ scheduler = Executors.newSingleThreadScheduledExecutor();
+ scheduler.scheduleAtFixedRate(this::scheduledTask, schedulerInitialDelay, schedulerPeriod, TimeUnit.SECONDS);
+ schedulerInitialized = true;
+ }
+ }
+ }
+ }
+
+ public void scheduledTask() {
+ this.stats = new SpRateLimiterStats();
+ stats.setAverageWaitTime(this.rateLimiterAverageWaitTime);
+ stats.setQueueSize(this.currentQueueSize.get());
+ stats.updateAllMetrics();
+ }
+
+ /**
+ * Creates a rate limiter with the specified parameters.
+ *
+ * @param permitsPerSecond The number of permits per second
+ * @param warmupPeriod The warmup period
+ * @param unit The time unit for the warmup period
+ * @throws IllegalArgumentException if parameters are invalid
+ */
+ public void createRateLimiter(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
+ if (this.rateLimiter == null) {
+ validateParameters(permitsPerSecond, warmupPeriod, unit);
+ this.rateLimiter = RateLimiter.create(permitsPerSecond, warmupPeriod, unit);
+ LOG.info("RateLimiter created with {} permits per second, warmup period: {} {}",
+ permitsPerSecond, warmupPeriod, unit);
+ } else {
+ LOG.warn("RateLimiter already exists. Use setRate() to modify the rate instead.");
+ }
+ }
+
+ /**
+ * Acquires a permit from the rate limiter for processing data, with timeout.
+ * Each request consumes exactly 1 permit regardless of data size.
+ * This provides simple and fair rate limiting based on request count.
+ *
+ * @param bytes The number of bytes to process (for logging purposes only)
+ * @return true if permit was acquired successfully, false if timeout occurred
+ * @throws InterruptedException if the current thread is interrupted while waiting
+ */
+ public boolean limit(long bytes) throws InterruptedException {
+ var timeOutMs = env.getRateLimiterTimeoutMs().getValueOrDefault();
+ if (this.rateLimiter == null) {
+ LOG.warn("RateLimiter has not been initialized. Please call createRateLimiter() first.");
+ return false;
+ }
+
+ long startTime = System.currentTimeMillis();
+
+ currentQueueSize.incrementAndGet();
+
+ try {
+ int permits = (int) bytes;
+ long timeoutMs = timeOutMs;
+ boolean acquired = rateLimiter.tryAcquire(permits, timeoutMs, TimeUnit.MILLISECONDS);
+
+ long waitTime = System.currentTimeMillis() - startTime;
+ updateAverageWaitTime(waitTime);
+
+ if (!acquired) {
+ LOG.warn("Failed to acquire permit for {} bytes within {} ms timeout (rate: {} requests/sec)",
+ bytes, timeoutMs, rateLimiter.getRate());
+ } else {
+ LOG.debug("Successfully acquired permit for {} bytes in {} ms (rate: {} requests/sec)",
+ bytes, waitTime, rateLimiter.getRate());
+ }
+ return acquired;
+ } finally {
+ currentQueueSize.updateAndGet(current -> Math.max(0, current - 1));
Review Comment:
decrementAndGet
--
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-unsubscribe@streampipes.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat:Add SpRateLimiter and SpMemoryManager
Posted by "luoluoyuyu (via GitHub)" <gi...@apache.org>.
luoluoyuyu commented on code in PR #3796:
URL: https://github.com/apache/streampipes/pull/3796#discussion_r2450167845
##########
streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/Limiter/SpRateLimiter.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.streampipes.extensions.api.Limiter;
+
+import com.google.common.util.concurrent.RateLimiter;
+import org.apache.streampipes.commons.prometheus.spRateLimiter.SpRateLimiterStats;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A singleton rate limiter implementation for StreamPipes extensions.
+ * This class provides rate limiting functionality using Google Guava's RateLimiter.
+ * It supports configurable permits per second and warmup periods.
+ */
+public enum SpRateLimiter {
+
+ INSTANCE;
+
+ private static final Logger LOG = LoggerFactory.getLogger(SpRateLimiter.class);
+
+ // Configuration constants
+ private static final double DEFAULT_PERMITS_PER_SECOND = 100.0;
+ private static final long DEFAULT_WARMUP_PERIOD = 1000L;
+ private static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS;
+ private static final int SCHEDULER_INITIAL_DELAY_SECONDS = 0;
+ private static final int SCHEDULER_PERIOD_SECONDS = 15;
+ private static final long ZERO_RATE_WAIT_TIME_MS = 1000L;
+ private static final int STATS_RESET_THRESHOLD = 1000;
+ private static final int STATS_RESET_FACTOR = 999;
+ private static final int STATS_RESET_DIVISOR = 1000;
+ private static final int SHUTDOWN_TIMEOUT_SECONDS = 5;
+
+ private RateLimiter rateLimiter;
+
+ private double rateLimiterQueueSize = 0.0;
+ private double rateLimiterAverageWaitTime = 0.0;
+
+ private long totalWaitTime = 0L;
+ private int waitTimeCount = 0;
+
+ private volatile int currentQueueSize = 0;
Review Comment:
Please use AtomicInteger type to complete the counting here
##########
streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/spRateLimiter/SpRateLimiterMetrics.java:
##########
@@ -0,0 +1,46 @@
+/*
+ * 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.streampipes.commons.prometheus.spRateLimiter;
+
+import io.prometheus.client.Gauge;
+import org.apache.streampipes.commons.prometheus.StreamPipesCollectorRegistry;
+
+/**
+ * Rate Limiter Metrics Manager
+ */
+public class SpRateLimiterMetrics {
+
+ public static final Gauge RATE_LIMITER_QUEUE_SIZE = StreamPipesCollectorRegistry.registerGauge(
+ "sp_rate_limiter_queue_size",
+ "Current size of the waiting queue"
+ );
+
+ public static final Gauge RATE_LIMITER_AVERAGE_WAIT_TIME = StreamPipesCollectorRegistry.registerGauge(
+ "sp_rate_limiter_average_wait_time_seconds",
+ "Average wait time for permit acquisition in seconds"
+ );
+
+ public static void updateCoreMetrics(double queueSize, double averageWaitTime) {
+ double safeQueueSize = Math.max(0.0, Math.min(queueSize, 10000.0));
+ double safeWaitTime = Math.max(0.0, Math.min(averageWaitTime, 3600.0));
Review Comment:
Shouldn't there be any need to set boundaries here?
##########
streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/Limiter/SpRateLimiter.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.streampipes.extensions.api.Limiter;
+
+import com.google.common.util.concurrent.RateLimiter;
+import org.apache.streampipes.commons.prometheus.spRateLimiter.SpRateLimiterStats;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A singleton rate limiter implementation for StreamPipes extensions.
+ * This class provides rate limiting functionality using Google Guava's RateLimiter.
+ * It supports configurable permits per second and warmup periods.
+ */
+public enum SpRateLimiter {
+
+ INSTANCE;
+
+ private static final Logger LOG = LoggerFactory.getLogger(SpRateLimiter.class);
+
+ // Configuration constants
+ private static final double DEFAULT_PERMITS_PER_SECOND = 100.0;
+ private static final long DEFAULT_WARMUP_PERIOD = 1000L;
+ private static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS;
+ private static final int SCHEDULER_INITIAL_DELAY_SECONDS = 0;
+ private static final int SCHEDULER_PERIOD_SECONDS = 15;
+ private static final long ZERO_RATE_WAIT_TIME_MS = 1000L;
+ private static final int STATS_RESET_THRESHOLD = 1000;
+ private static final int STATS_RESET_FACTOR = 999;
+ private static final int STATS_RESET_DIVISOR = 1000;
+ private static final int SHUTDOWN_TIMEOUT_SECONDS = 5;
+
+ private RateLimiter rateLimiter;
+
+ private double rateLimiterQueueSize = 0.0;
+ private double rateLimiterAverageWaitTime = 0.0;
+
+ private long totalWaitTime = 0L;
+ private int waitTimeCount = 0;
+
+ private volatile int currentQueueSize = 0;
+
+ private SpRateLimiterStats stats;
+ private static volatile boolean schedulerInitialized = false;
+ private static ScheduledExecutorService scheduler;
+
+ /**
+ * Creates a rate limiter with default parameters.
+ * Default: 100 permits per second, 1000ms warmup period.
+ */
+ public void createRateLimiter() {
+ createRateLimiter(DEFAULT_PERMITS_PER_SECOND, DEFAULT_WARMUP_PERIOD, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created and scheduler initialized");
+ }
+
+ /**
+ * Creates a rate limiter with the specified permits per second.
+ * Uses default warmup period of 1000ms.
+ *
+ * @param permitsPerSecond The number of permits per second
+ */
+ public void createRateLimiter(double permitsPerSecond) {
+ createRateLimiter(permitsPerSecond, DEFAULT_WARMUP_PERIOD, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created with {} permits/sec and scheduler initialized", permitsPerSecond);
+ }
+
+ public void initScheduledTasks() {
+ if (!schedulerInitialized) {
+ synchronized (SpRateLimiter.class) {
+ if (!schedulerInitialized) {
+ scheduler = Executors.newSingleThreadScheduledExecutor();
+ scheduler.scheduleAtFixedRate(this::ScheduledTask, SCHEDULER_INITIAL_DELAY_SECONDS, SCHEDULER_PERIOD_SECONDS, TimeUnit.SECONDS);
+ schedulerInitialized = true;
+ }
+ }
+ }
+ }
+
+ public void ScheduledTask() {
+ this.stats = new SpRateLimiterStats();
+ stats.setAverageWaitTime(this.rateLimiterAverageWaitTime);
+ stats.setQueueSize(this.rateLimiterQueueSize);
+ stats.updateAllMetrics();
+ }
+
+ /**
+ * Creates a rate limiter with the specified parameters.
+ *
+ * @param permitsPerSecond The number of permits per second
+ * @param warmupPeriod The warmup period
+ * @param unit The time unit for the warmup period
+ * @throws IllegalArgumentException if parameters are invalid
+ */
+ public void createRateLimiter(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
+ if (this.rateLimiter == null) {
+ validateParameters(permitsPerSecond, warmupPeriod, unit);
+ this.rateLimiter = RateLimiter.create(permitsPerSecond, warmupPeriod, unit);
+ LOG.info("RateLimiter created with {} permits per second, warmup period: {} {}",
+ permitsPerSecond, warmupPeriod, unit);
+ } else {
+ LOG.warn("RateLimiter already exists. Use setRate() to modify the rate instead.");
+ }
+ }
+
+ /**
+ * Acquires a permit from the rate limiter, blocking if necessary.
+ * If the rate limiter is not initialized, logs a warning and returns immediately.
+ * If the rate is zero or negative, logs a warning and sleeps for 1 second.
+ *
+ * @throws InterruptedException if the current thread is interrupted while waiting
+ */
+ public void limit() throws InterruptedException {
+ if (this.rateLimiter == null) {
+ LOG.warn("RateLimiter has not been initialized. Please call createRateLimiter() first.");
+ return;
+ }
+
+ long startTime = System.currentTimeMillis();
+
+ synchronized (this) {
+ currentQueueSize++;
+ rateLimiterQueueSize = currentQueueSize;
+ }
+
+ try {
+ if (rateLimiter.getRate() <= 0) {
+ LOG.warn("RateLimiter is set to zero or negative rate. No permits will be acquired.");
+ updateAverageWaitTime(ZERO_RATE_WAIT_TIME_MS);
+ Thread.sleep(ZERO_RATE_WAIT_TIME_MS);
+ } else {
+ this.rateLimiter.acquire();
+ long waitTime = System.currentTimeMillis() - startTime;
Review Comment:
The Rate here should be used to limit the speed limit, so it is the value set at initialization, and whether this function ```tryAcquire(int permits, long timeout, TimeUnit unit)``` should be used. Permits is the byte length at runtime.
##########
streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/spMemoryManager/SpMemoryManagerMetrics.java:
##########
@@ -0,0 +1,46 @@
+/*
+ * 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.streampipes.commons.prometheus.spMemoryManager;
+
+import io.prometheus.client.Gauge;
+import org.apache.streampipes.commons.prometheus.StreamPipesCollectorRegistry;
+
+/**
+ * Memory Manager Metrics Manager
+ */
+public class SpMemoryManagerMetrics {
+
+ public static final Gauge MEMORY_USED_BYTES = StreamPipesCollectorRegistry.registerGauge(
+ "sp_memory_used_bytes",
+ "Amount of memory used in bytes"
+ );
+
+ public static final Gauge MEMORY_ALLOCATION_RATE = StreamPipesCollectorRegistry.registerGauge(
+ "sp_memory_allocation_rate_bytes_per_second",
+ "Memory allocation rate in bytes per second"
+ );
+
+ public static void updateCoreMetrics(double usedMemory, double totalMemory, double allocationRate) {
+ double safeUsedMemory = Math.max(0.0, Math.min(usedMemory, totalMemory));
+ double safeAllocationRate = Math.max(0.0, Math.min(allocationRate, 1e12));
Review Comment:
Please check whether the logic here is usedMemory/totalMemory, and whether there are any problems with other logic
##########
streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/Limiter/SpRateLimiter.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.streampipes.extensions.api.Limiter;
+
+import com.google.common.util.concurrent.RateLimiter;
+import org.apache.streampipes.commons.prometheus.spRateLimiter.SpRateLimiterStats;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A singleton rate limiter implementation for StreamPipes extensions.
+ * This class provides rate limiting functionality using Google Guava's RateLimiter.
+ * It supports configurable permits per second and warmup periods.
+ */
+public enum SpRateLimiter {
+
+ INSTANCE;
+
+ private static final Logger LOG = LoggerFactory.getLogger(SpRateLimiter.class);
+
+ // Configuration constants
+ private static final double DEFAULT_PERMITS_PER_SECOND = 100.0;
+ private static final long DEFAULT_WARMUP_PERIOD = 1000L;
+ private static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS;
+ private static final int SCHEDULER_INITIAL_DELAY_SECONDS = 0;
+ private static final int SCHEDULER_PERIOD_SECONDS = 15;
+ private static final long ZERO_RATE_WAIT_TIME_MS = 1000L;
+ private static final int STATS_RESET_THRESHOLD = 1000;
+ private static final int STATS_RESET_FACTOR = 999;
+ private static final int STATS_RESET_DIVISOR = 1000;
+ private static final int SHUTDOWN_TIMEOUT_SECONDS = 5;
+
+ private RateLimiter rateLimiter;
+
+ private double rateLimiterQueueSize = 0.0;
+ private double rateLimiterAverageWaitTime = 0.0;
+
+ private long totalWaitTime = 0L;
+ private int waitTimeCount = 0;
+
+ private volatile int currentQueueSize = 0;
+
+ private SpRateLimiterStats stats;
+ private static volatile boolean schedulerInitialized = false;
+ private static ScheduledExecutorService scheduler;
+
+ /**
+ * Creates a rate limiter with default parameters.
+ * Default: 100 permits per second, 1000ms warmup period.
+ */
+ public void createRateLimiter() {
+ createRateLimiter(DEFAULT_PERMITS_PER_SECOND, DEFAULT_WARMUP_PERIOD, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created and scheduler initialized");
+ }
+
+ /**
+ * Creates a rate limiter with the specified permits per second.
+ * Uses default warmup period of 1000ms.
+ *
+ * @param permitsPerSecond The number of permits per second
+ */
+ public void createRateLimiter(double permitsPerSecond) {
+ createRateLimiter(permitsPerSecond, DEFAULT_WARMUP_PERIOD, DEFAULT_TIME_UNIT);
+ initScheduledTasks();
+ LOG.info("RateLimiter created with {} permits/sec and scheduler initialized", permitsPerSecond);
+ }
+
+ public void initScheduledTasks() {
+ if (!schedulerInitialized) {
+ synchronized (SpRateLimiter.class) {
+ if (!schedulerInitialized) {
+ scheduler = Executors.newSingleThreadScheduledExecutor();
+ scheduler.scheduleAtFixedRate(this::ScheduledTask, SCHEDULER_INITIAL_DELAY_SECONDS, SCHEDULER_PERIOD_SECONDS, TimeUnit.SECONDS);
+ schedulerInitialized = true;
+ }
+ }
+ }
+ }
+
+ public void ScheduledTask() {
+ this.stats = new SpRateLimiterStats();
+ stats.setAverageWaitTime(this.rateLimiterAverageWaitTime);
+ stats.setQueueSize(this.rateLimiterQueueSize);
+ stats.updateAllMetrics();
+ }
+
+ /**
+ * Creates a rate limiter with the specified parameters.
+ *
+ * @param permitsPerSecond The number of permits per second
+ * @param warmupPeriod The warmup period
+ * @param unit The time unit for the warmup period
+ * @throws IllegalArgumentException if parameters are invalid
+ */
+ public void createRateLimiter(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
+ if (this.rateLimiter == null) {
+ validateParameters(permitsPerSecond, warmupPeriod, unit);
+ this.rateLimiter = RateLimiter.create(permitsPerSecond, warmupPeriod, unit);
+ LOG.info("RateLimiter created with {} permits per second, warmup period: {} {}",
+ permitsPerSecond, warmupPeriod, unit);
+ } else {
+ LOG.warn("RateLimiter already exists. Use setRate() to modify the rate instead.");
+ }
+ }
+
+ /**
+ * Acquires a permit from the rate limiter, blocking if necessary.
+ * If the rate limiter is not initialized, logs a warning and returns immediately.
+ * If the rate is zero or negative, logs a warning and sleeps for 1 second.
+ *
+ * @throws InterruptedException if the current thread is interrupted while waiting
+ */
+ public void limit() throws InterruptedException {
+ if (this.rateLimiter == null) {
+ LOG.warn("RateLimiter has not been initialized. Please call createRateLimiter() first.");
+ return;
+ }
+
+ long startTime = System.currentTimeMillis();
+
+ synchronized (this) {
+ currentQueueSize++;
+ rateLimiterQueueSize = currentQueueSize;
Review Comment:
Should only one variable be used to calculate the size? The other one might be redundant.
##########
streampipes-messaging-jms/src/main/java/org/apache/streampipes/messaging/jms/ActiveMQConsumer.java:
##########
@@ -52,7 +52,11 @@ private void initListener() {
consumer.setMessageListener(message -> {
if (message instanceof BytesMessage) {
ByteSequence bs = ((ActiveMQBytesMessage) message).getContent();
- eventProcessor.onEvent(bs.getData());
+ try {
+ eventProcessor.onEvent(bs.getData());
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
Review Comment:
Please refer to ```https://docs.oracle.com/javase/tutorial/essential/concurrency/interrupt.html``` to handle interrupt exceptions.
--
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-unsubscribe@streampipes.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat:Add SpRateLimiter and SpMemoryManager
Posted by "luoluoyuyu (via GitHub)" <gi...@apache.org>.
luoluoyuyu commented on PR #3796:
URL: https://github.com/apache/streampipes/pull/3796#issuecomment-3456917934
@cursor review
--
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-unsubscribe@streampipes.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat:Add SpRateLimiter and SpMemoryManager
Posted by "luoluoyuyu (via GitHub)" <gi...@apache.org>.
luoluoyuyu merged PR #3796:
URL: https://github.com/apache/streampipes/pull/3796
--
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-unsubscribe@streampipes.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org