Skip to content

[feat][broker]: intercept transaction begin and end events for observability #14613

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
codelipenghui merged 7 commits into from
Mar 18, 2022
Merged

[feat][broker]: intercept transaction begin and end events for observability #14613

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
c7dff9f
Interceptors for transaction begin and end events
madhavan-narayanan Jan 11, 2022
87c37ff
Merge branch 'apache:master' into master
madhavan-narayanan Jan 15, 2022
683b766
Merge branch 'apache:master' into master
madhavan-narayanan Jan 31, 2022
7371a70
Merge branch 'apache:master' into master
madhavan-narayanan Mar 9, 2022
c833b63
Merge branch 'apache:master' into master
madhavan-narayanan Mar 16, 2022
9a8529e
Merge branch 'apache:master' into master
madhavan-narayanan Mar 17, 2022
50243b7
pulsar-broker: Changes to address review comments in PR#14613
madhavan-narayanan Mar 17, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,23 @@ default void messageAcked(ServerCnx cnx, Consumer consumer,
CommandAck ackCmd) {
}

/**
* Intercept when a transaction begins.
*
* @param tcId Transaction Coordinator Id
* @param txnID Transaction ID
*/
default void txnOpened(long tcId, String txnID) {
}

/**
* Intercept when a transaction ends.
*
* @param txnID Transaction ID
* @param txnAction Transaction Action
*/
default void txnEnded(String txnID, long txnAction) {
}
/**
* Called by the broker while new command incoming.
*/
Expand Down
10 changes: 10 additions & 0 deletions ...er/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,16 @@ public void messageAcked(ServerCnx cnx, Consumer consumer,
}
}

@Override
public void txnOpened(long tcId, String txnID) {
this.interceptor.txnOpened(tcId, txnID);
}

@Override
public void txnEnded(String txnID, long txnAction) {
this.interceptor.txnEnded(txnID, txnAction);
}

@Override
public void onConnectionCreated(ServerCnx cnx) {
try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) {
Expand Down
21 changes: 21 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,27 @@ public void messageAcked(ServerCnx cnx, Consumer consumer,
}
}

@Override
public void txnOpened(long tcId, String txnID) {
if (interceptors == null || interceptors.isEmpty()) {
return;
}
for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
value.txnOpened(tcId, txnID);
}
}

@Override
public void txnEnded(String txnID, long txnAction) {
if (interceptors == null || interceptors.isEmpty()) {
return;
}
for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
value.txnEnded(txnID, txnAction);
}
}


@Override
public void onConnectionCreated(ServerCnx cnx) {
if (interceptors == null || interceptors.isEmpty()) {
Expand Down
8 changes: 8 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import java.util.List;
import java.util.Optional;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.common.api.proto.CommandLookupTopicResponse;
import org.apache.pulsar.common.api.proto.ServerError;
import org.apache.pulsar.common.protocol.schema.SchemaVersion;
Expand Down Expand Up @@ -83,4 +84,11 @@ Future<Void> sendMessagesToConsumer(long consumerId, String topicName, Subscript

void sendTcClientConnectResponse(long requestId);

void sendNewTxnResponse(long requestId, TxnID txnID, long tcID);

void sendNewTxnErrorResponse(long requestId, long txnID, ServerError error, String message);

void sendEndTxnResponse(long requestId, TxnID txnID, int txnAction);

void sendEndTxnErrorResponse(long requestId, TxnID txnID, ServerError error, String message);
}
45 changes: 45 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,10 +28,12 @@
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.common.api.proto.BaseCommand;
import org.apache.pulsar.common.api.proto.CommandLookupTopicResponse;
import org.apache.pulsar.common.api.proto.ProtocolVersion;
import org.apache.pulsar.common.api.proto.ServerError;
import org.apache.pulsar.common.api.proto.TxnAction;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.protocol.schema.SchemaVersion;
import org.apache.pulsar.common.schema.SchemaInfo;
Expand Down Expand Up @@ -299,6 +301,49 @@ public void sendTcClientConnectResponse(long requestId) {
sendTcClientConnectResponse(requestId, null, null);
}

@Override
public void sendNewTxnResponse(long requestId, TxnID txnID, long tcID) {
BaseCommand command = Commands.newTxnResponse(requestId, txnID.getLeastSigBits(),
txnID.getMostSigBits());
safeIntercept(command, cnx);
ByteBuf outBuf = Commands.serializeWithSize(command);
cnx.ctx().writeAndFlush(outBuf);
if (this.interceptor != null) {
this.interceptor.txnOpened(tcID, txnID.toString());
}
}

@Override
public void sendNewTxnErrorResponse(long requestId, long txnID, ServerError error, String message) {
BaseCommand command = Commands.newTxnResponse(requestId, txnID, error, message);
safeIntercept(command, cnx);
ByteBuf outBuf = Commands.serializeWithSize(command);
cnx.ctx().writeAndFlush(outBuf);
}

@Override
public void sendEndTxnResponse(long requestId, TxnID txnID, int txnAction) {
BaseCommand command = Commands.newEndTxnResponse(requestId, txnID.getLeastSigBits(),
txnID.getMostSigBits());
safeIntercept(command, cnx);
ByteBuf outBuf = Commands.serializeWithSize(command);
cnx.ctx().writeAndFlush(outBuf);
if (this.interceptor != null) {
this.interceptor.txnEnded(txnID.toString(), txnAction);
}
}

@Override
public void sendEndTxnErrorResponse(long requestId, TxnID txnID, ServerError error, String message) {
BaseCommand command = Commands.newEndTxnResponse(requestId, txnID.getMostSigBits(), error, message);
safeIntercept(command, cnx);
ByteBuf outBuf = Commands.serializeWithSize(command);
cnx.ctx().writeAndFlush(outBuf);
if (this.interceptor != null) {
this.interceptor.txnEnded(txnID.toString(), TxnAction.ABORT_VALUE);
}
}

private void safeIntercept(BaseCommand command, ServerCnx cnx) {
try {
this.interceptor.onPulsarCommand(command, cnx);
Expand Down
15 changes: 7 additions & 8 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2175,13 +2175,12 @@ protected void handleNewTxn(CommandNewTxn command) {
if (log.isDebugEnabled()) {
log.debug("Send response {} for new txn request {}", tcId.getId(), requestId);
}
ctx.writeAndFlush(Commands.newTxnResponse(requestId, txnID.getLeastSigBits(),
txnID.getMostSigBits()));
commandSender.sendNewTxnResponse(requestId, txnID, command.getTcId());
} else {
ex = handleTxnException(ex, BaseCommand.Type.NEW_TXN.name(), requestId);

ctx.writeAndFlush(Commands.newTxnResponse(requestId, tcId.getId(),
BrokerServiceException.getClientErrorCode(ex), ex.getMessage()));
commandSender.sendNewTxnErrorResponse(requestId, tcId.getId(),
BrokerServiceException.getClientErrorCode(ex), ex.getMessage());
transactionMetadataStoreService.handleOpFail(ex, tcId);
}
}));
Expand Down Expand Up @@ -2242,12 +2241,12 @@ protected void handleEndTxn(CommandEndTxn command) {
.endTransaction(txnID, txnAction, false)
.whenComplete((v, ex) -> {
if (ex == null) {
ctx.writeAndFlush(Commands.newEndTxnResponse(requestId,
txnID.getLeastSigBits(), txnID.getMostSigBits()));
commandSender.sendEndTxnResponse(requestId,
txnID, txnAction);
} else {
ex = handleTxnException(ex, BaseCommand.Type.END_TXN.name(), requestId);
ctx.writeAndFlush(Commands.newEndTxnResponse(requestId, txnID.getMostSigBits(),
BrokerServiceException.getClientErrorCode(ex), ex.getMessage()));
commandSender.sendEndTxnErrorResponse(requestId, txnID,
BrokerServiceException.getClientErrorCode(ex), ex.getMessage());

transactionMetadataStoreService.handleOpFail(ex, tcId);
}
Expand Down
1 change: 1 addition & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2949,6 +2949,7 @@ public void publishTxnMessage(TxnID txnID, ByteBuf headersAndPayload, PublishCon
// Message has been successfully persisted
messageDeduplication.recordMessagePersisted(publishContext,
(PositionImpl) position);
publishContext.setProperty("txn_id", txnID.toString());
publishContext.completed(null, ((PositionImpl) position).getLedgerId(),
((PositionImpl) position).getEntryId());

Expand Down
30 changes: 30 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
import org.apache.pulsar.common.api.proto.BaseCommand;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.api.proto.CommandAck;
import org.apache.pulsar.common.api.proto.TxnAction;
import org.eclipse.jetty.server.Response;


Expand All @@ -56,6 +57,9 @@ public class CounterBrokerInterceptor implements BrokerInterceptor {
int messageDispatchCount = 0;
int messageAckCount = 0;
int handleAckCount = 0;
int txnCount = 0;
int committedTxnCount = 0;
int abortedTxnCount = 0;

private List<ResponseEvent> responseList = new ArrayList<>();

Expand Down Expand Up @@ -177,6 +181,20 @@ public void onFilter(ServletRequest request, ServletResponse response, FilterCha
chain.doFilter(request, response);
}

@Override
public void txnOpened(long tcId, String txnID) {
txnCount ++;
}

@Override
public void txnEnded(String txnID, long txnAction) {
if(txnAction == TxnAction.COMMIT_VALUE) {
committedTxnCount ++;
} else {
abortedTxnCount ++;
}
}

@Override
public void initialize(PulsarService pulsarService) throws Exception {

Expand Down Expand Up @@ -230,4 +248,16 @@ public void clearResponseList() {
public List<ResponseEvent> getResponseList() {
return responseList;
}

public int getTxnCount() {
return txnCount;
}

public int getCommittedTxnCount() {
return committedTxnCount;
}

public int getAbortedTxnCount() {
return abortedTxnCount;
}
}
10 changes: 8 additions & 2 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,8 @@
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.intercept.CounterBrokerInterceptor;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
Expand Down Expand Up @@ -699,8 +701,12 @@ public void testEndTxnWhenCommittingOrAborting() throws Exception {
field.set(commitTxn, TransactionImpl.State.COMMITTING);
field.set(abortTxn, TransactionImpl.State.ABORTING);

abortTxn.abort();
commitTxn.commit();
BrokerInterceptor listener = getPulsarServiceList().get(0).getBrokerInterceptor();
assertEquals(((CounterBrokerInterceptor)listener).getTxnCount(),2);
abortTxn.abort().get();
assertEquals(((CounterBrokerInterceptor)listener).getAbortedTxnCount(),1);
commitTxn.commit().get();
assertEquals(((CounterBrokerInterceptor)listener).getCommittedTxnCount(),1);
}

@Test
Expand Down
17 changes: 9 additions & 8 deletions pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1264,16 +1264,16 @@ public static ByteBuf newTxn(long tcId, long requestId, long ttlSeconds) {
return serializeWithSize(cmd);
}

public static ByteBuf newTxnResponse(long requestId, long txnIdLeastBits, long txnIdMostBits) {
public static BaseCommand newTxnResponse(long requestId, long txnIdLeastBits, long txnIdMostBits) {
BaseCommand cmd = localCmd(Type.NEW_TXN_RESPONSE);
cmd.setNewTxnResponse()
.setRequestId(requestId)
.setTxnidMostBits(txnIdMostBits)
.setTxnidLeastBits(txnIdLeastBits);
return serializeWithSize(cmd);
return cmd;
}

public static ByteBuf newTxnResponse(long requestId, long txnIdMostBits, ServerError error, String errorMsg) {
public static BaseCommand newTxnResponse(long requestId, long txnIdMostBits, ServerError error, String errorMsg) {
BaseCommand cmd = localCmd(Type.NEW_TXN_RESPONSE);
CommandNewTxnResponse response = cmd.setNewTxnResponse()
.setRequestId(requestId)
Expand All @@ -1282,7 +1282,7 @@ public static ByteBuf newTxnResponse(long requestId, long txnIdMostBits, ServerE
if (errorMsg != null) {
response.setMessage(errorMsg);
}
return serializeWithSize(cmd);
return cmd;
}

public static ByteBuf newAddPartitionToTxn(long requestId, long txnIdLeastBits, long txnIdMostBits,
Expand Down Expand Up @@ -1363,16 +1363,17 @@ public static BaseCommand newEndTxn(long requestId, long txnIdLeastBits, long tx
return cmd;
}

public static ByteBuf newEndTxnResponse(long requestId, long txnIdLeastBits, long txnIdMostBits) {
public static BaseCommand newEndTxnResponse(long requestId, long txnIdLeastBits, long txnIdMostBits) {
BaseCommand cmd = localCmd(Type.END_TXN_RESPONSE);
cmd.setEndTxnResponse()
.setRequestId(requestId)
.setTxnidLeastBits(txnIdLeastBits)
.setTxnidMostBits(txnIdMostBits);
return serializeWithSize(cmd);
return cmd;
}

public static ByteBuf newEndTxnResponse(long requestId, long txnIdMostBits, ServerError error, String errorMsg) {
public static BaseCommand newEndTxnResponse(long requestId, long txnIdMostBits,
ServerError error, String errorMsg) {
BaseCommand cmd = localCmd(Type.END_TXN_RESPONSE);
CommandEndTxnResponse response = cmd.setEndTxnResponse()
.setRequestId(requestId)
Expand All @@ -1381,7 +1382,7 @@ public static ByteBuf newEndTxnResponse(long requestId, long txnIdMostBits, Serv
if (errorMsg != null) {
response.setMessage(errorMsg);
}
return serializeWithSize(cmd);
return cmd;
}

public static ByteBuf newEndTxnOnPartition(long requestId, long txnIdLeastBits, long txnIdMostBits, String topic,
Expand Down