You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/04/27 15:23:09 UTC
[GitHub] [pulsar] congbobo184 opened a new pull request #10407: [Transaction] Fix transaction buffer client channel not active problem.
congbobo184 opened a new pull request #10407:
URL: https://github.com/apache/pulsar/pull/10407
## Motivation
when channel is not active, it will not invalidate the `ClientCnx`.
## implement
check channel active.
### Verifying this change
Add the tests for it
Does this pull request potentially affect one of the following parts:
If yes was chosen, please highlight the changes
Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10407: [Transaction] Fix transaction buffer client channel not active problem.
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10407:
URL: https://github.com/apache/pulsar/pull/10407#discussion_r622227403
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferHandlerImpl.java
##########
@@ -116,24 +124,26 @@ public TransactionBufferHandlerImpl(PulsarClient pulsarClient,
private CompletableFuture<TxnID> endTxn(long requestId, String topic, ByteBuf cmd, CompletableFuture<TxnID> cb) {
OpRequestSend op = OpRequestSend.create(requestId, topic, cmd, cb);
- pendingRequests.put(requestId, op);
- cmd.retain();
try {
cache.get(topic).whenComplete((clientCnx, throwable) -> {
if (throwable == null) {
- try {
+ if (clientCnx.ctx().channel().isActive()) {
clientCnx.registerTransactionBufferHandler(TransactionBufferHandlerImpl.this);
+ synchronized (TransactionBufferHandlerImpl.class) {
Review comment:
It seems odd to use the Class object as an object monitor. Is this intentional?
Did you mean to have `synchronized (TransactionBufferHandlerImpl.this) {` instead?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli merged pull request #10407: [Transaction] Fix transaction buffer client channel not active problem.
Posted by GitBox <gi...@apache.org>.
eolivelli merged pull request #10407:
URL: https://github.com/apache/pulsar/pull/10407
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #10407: [Transaction] Fix transaction buffer client channel not active problem.
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #10407:
URL: https://github.com/apache/pulsar/pull/10407#discussion_r622023271
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferHandlerImpl.java
##########
@@ -116,24 +124,26 @@ public TransactionBufferHandlerImpl(PulsarClient pulsarClient,
private CompletableFuture<TxnID> endTxn(long requestId, String topic, ByteBuf cmd, CompletableFuture<TxnID> cb) {
OpRequestSend op = OpRequestSend.create(requestId, topic, cmd, cb);
- pendingRequests.put(requestId, op);
- cmd.retain();
try {
cache.get(topic).whenComplete((clientCnx, throwable) -> {
if (throwable == null) {
- try {
+ if (clientCnx.ctx().channel().isActive()) {
clientCnx.registerTransactionBufferHandler(TransactionBufferHandlerImpl.this);
+ synchronized (TransactionBufferHandlerImpl.class) {
+ pendingRequests.put(requestId, op);
+ cmd.retain();
+ }
clientCnx.ctx().writeAndFlush(cmd, clientCnx.ctx().voidPromise());
- } catch (Exception e) {
+ } else {
cache.invalidate(topic);
- cb.completeExceptionally(e);
- pendingRequests.remove(requestId);
+ cb.completeExceptionally(
+ new PulsarClientException.LookupException(topic + " endTxn channel is not active"));
op.recycle();
}
} else {
+ log.error("endTxn error topic: [{}]", topic, throwable);
cache.invalidate(topic);
cb.completeExceptionally(throwable);
- pendingRequests.remove(requestId);
Review comment:
Why are you removing this line?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #10407: [Transaction] Fix transaction buffer client channel not active problem.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #10407:
URL: https://github.com/apache/pulsar/pull/10407#discussion_r622232539
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferHandlerImpl.java
##########
@@ -116,24 +124,26 @@ public TransactionBufferHandlerImpl(PulsarClient pulsarClient,
private CompletableFuture<TxnID> endTxn(long requestId, String topic, ByteBuf cmd, CompletableFuture<TxnID> cb) {
OpRequestSend op = OpRequestSend.create(requestId, topic, cmd, cb);
- pendingRequests.put(requestId, op);
- cmd.retain();
try {
cache.get(topic).whenComplete((clientCnx, throwable) -> {
if (throwable == null) {
- try {
+ if (clientCnx.ctx().channel().isActive()) {
clientCnx.registerTransactionBufferHandler(TransactionBufferHandlerImpl.this);
+ synchronized (TransactionBufferHandlerImpl.class) {
Review comment:
it my mistake, good catch!
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #10407: [Transaction] Fix transaction buffer client channel not active problem.
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #10407:
URL: https://github.com/apache/pulsar/pull/10407#discussion_r622191046
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferHandlerImpl.java
##########
@@ -116,24 +124,26 @@ public TransactionBufferHandlerImpl(PulsarClient pulsarClient,
private CompletableFuture<TxnID> endTxn(long requestId, String topic, ByteBuf cmd, CompletableFuture<TxnID> cb) {
OpRequestSend op = OpRequestSend.create(requestId, topic, cmd, cb);
- pendingRequests.put(requestId, op);
- cmd.retain();
try {
cache.get(topic).whenComplete((clientCnx, throwable) -> {
if (throwable == null) {
- try {
+ if (clientCnx.ctx().channel().isActive()) {
clientCnx.registerTransactionBufferHandler(TransactionBufferHandlerImpl.this);
+ synchronized (TransactionBufferHandlerImpl.class) {
+ pendingRequests.put(requestId, op);
+ cmd.retain();
+ }
clientCnx.ctx().writeAndFlush(cmd, clientCnx.ctx().voidPromise());
- } catch (Exception e) {
+ } else {
cache.invalidate(topic);
- cb.completeExceptionally(e);
- pendingRequests.remove(requestId);
+ cb.completeExceptionally(
+ new PulsarClientException.LookupException(topic + " endTxn channel is not active"));
op.recycle();
}
} else {
+ log.error("endTxn error topic: [{}]", topic, throwable);
cache.invalidate(topic);
cb.completeExceptionally(throwable);
- pendingRequests.remove(requestId);
Review comment:
makes sense, thanks
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #10407: [Transaction] Fix transaction buffer client channel not active problem.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #10407:
URL: https://github.com/apache/pulsar/pull/10407#discussion_r622128765
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferHandlerImpl.java
##########
@@ -116,24 +124,26 @@ public TransactionBufferHandlerImpl(PulsarClient pulsarClient,
private CompletableFuture<TxnID> endTxn(long requestId, String topic, ByteBuf cmd, CompletableFuture<TxnID> cb) {
OpRequestSend op = OpRequestSend.create(requestId, topic, cmd, cb);
- pendingRequests.put(requestId, op);
- cmd.retain();
try {
cache.get(topic).whenComplete((clientCnx, throwable) -> {
if (throwable == null) {
- try {
+ if (clientCnx.ctx().channel().isActive()) {
clientCnx.registerTransactionBufferHandler(TransactionBufferHandlerImpl.this);
+ synchronized (TransactionBufferHandlerImpl.class) {
+ pendingRequests.put(requestId, op);
+ cmd.retain();
+ }
clientCnx.ctx().writeAndFlush(cmd, clientCnx.ctx().voidPromise());
- } catch (Exception e) {
+ } else {
cache.invalidate(topic);
- cb.completeExceptionally(e);
- pendingRequests.remove(requestId);
+ cb.completeExceptionally(
+ new PulsarClientException.LookupException(topic + " endTxn channel is not active"));
op.recycle();
}
} else {
+ log.error("endTxn error topic: [{}]", topic, throwable);
cache.invalidate(topic);
cb.completeExceptionally(throwable);
- pendingRequests.remove(requestId);
Review comment:
if run here, the request don't put in `pendingRequests`. so remove this line
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org