Add log error tracking for semaphore count leak
@@ -148,6 +148,8 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
private Optional<Long> topicEpoch = Optional.empty();
private final List<Throwable> previousExceptions = new CopyOnWriteArrayList<Throwable>();
+ private boolean errorState;
+
@SuppressWarnings("rawtypes")
private static final AtomicLongFieldUpdater<ProducerImpl> msgIdGeneratorUpdater = AtomicLongFieldUpdater
.newUpdater(ProducerImpl.class, "msgIdGenerator");
@@ -258,6 +260,21 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
grabCnx();
}
+ protected void semaphoreRelease(final int releaseCountRequest) {
+ if (semaphore.isPresent()) {
+ if (!errorState) {
+ final int availablePermits = semaphore.get().availablePermits();
+ if (availablePermits - releaseCountRequest < 0) {
+ log.error("Semaphore permit release count request greater then availablePermits" +
+ " : availablePermits={}, releaseCountRequest={}",
+ availablePermits, releaseCountRequest);
+ errorState = true;
+ }
+ }
+ semaphore.get().release(releaseCountRequest);
+ }
+ }
+
protected OpSendMsgQueue createPendingMessagesQueue() {
return new OpSendMsgQueue();
}
@@ -1005,9 +1022,9 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
}
private void releaseSemaphoreForSendOp(OpSendMsg op) {
- if (semaphore.isPresent()) {
- semaphore.get().release(isBatchMessagingEnabled() ? op.numMessagesInBatch : 1);
- }
+
+ semaphoreRelease(isBatchMessagingEnabled() ? op.numMessagesInBatch : 1);
+
client.getMemoryLimitController().releaseMemory(op.uncompressedSize);
}
@@ -1706,7 +1723,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
});
pendingMessages.clear();
- semaphore.ifPresent(s -> s.release(releaseCount.get()));
+ semaphoreRelease(releaseCount.get());
if (batchMessagingEnabled) {
failPendingBatchMessages(ex);
}
@@ -1732,7 +1749,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
}
final int numMessagesInBatch = batchMessageContainer.getNumMessagesInBatch();
batchMessageContainer.discard(ex);
- semaphore.ifPresent(s -> s.release(numMessagesInBatch));
+ semaphoreRelease(numMessagesInBatch);
}
@Override
@@ -1776,9 +1793,9 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
}
} catch (PulsarClientException e) {
Thread.currentThread().interrupt();
- semaphore.ifPresent(s -> s.release(batchMessageContainer.getNumMessagesInBatch()));
+ semaphoreRelease(batchMessageContainer.getNumMessagesInBatch());
} catch (Throwable t) {
- semaphore.ifPresent(s -> s.release(batchMessageContainer.getNumMessagesInBatch()));
+ semaphoreRelease(batchMessageContainer.getNumMessagesInBatch());
log.warn("[{}] [{}] error while create opSendMsg by batch message container", topic, producerName, t);
}
}
