Skip to content

[fix][broker] Fix messages could not expire due to incorrect client clock #21835

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

Closed
315157973 wants to merge 8 commits into from
+26 −1
Closed

[fix][broker] Fix messages could not expire due to incorrect client clock #21835

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
2fdf4c8
[fix][test]Fix resource not close after method
315157973 Jul 4, 2023
b21dc48
Merge pull request #11 from 315157973/release
315157973 Jul 5, 2023
8f0dbd3
Merge branch 'master' of github.com:315157973/pulsar
315157973 Jul 10, 2023
4006265
Merge remote-tracking branch 'upstream/master'
315157973 Dec 28, 2023
33e9158
[fix][broker] Fix messages could not expire due to incorrect client c…
315157973 Jan 2, 2024
b9bf55a
[fix][broker] Fix messages could not expire due to incorrect client c…
315157973 Jan 3, 2024
523ac76
Address comment
315157973 Jan 4, 2024
2f97734
merge master
315157973 Aug 14, 2024
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
5 changes: 4 additions & 1 deletion 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 @@ -472,7 +472,7 @@ public static void skipMessageMetadata(ByteBuf buffer) {
}

public static long getEntryTimestamp(ByteBuf headersAndPayloadWithBrokerEntryMetadata) throws IOException {
// get broker timestamp first if BrokerEntryMetadata is enabled with AppendBrokerTimestampMetadataInterceptor
// get broker timestamp first if exists
BrokerEntryMetadata brokerEntryMetadata =
Commands.parseBrokerEntryMetadataIfExist(headersAndPayloadWithBrokerEntryMetadata);
if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
Expand Down Expand Up @@ -1716,6 +1716,9 @@ public static ByteBuf addBrokerEntryMetadata(ByteBuf headerAndPayload,
interceptor.interceptWithNumberOfMessages(brokerEntryMetadata, numberOfMessages);
}
}
if (!brokerEntryMetadata.hasBrokerTimestamp()) {
brokerEntryMetadata.setBrokerTimestamp(System.currentTimeMillis());
}
Comment on lines +1719 to +1721

@codelipenghui codelipenghui Jan 4, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The broker timestamp will be set in line:1702 if the AppendBrokerTimestampMetadataInterceptor is enabled. Is this PR wants to set the broker timestamp without AppendBrokerTimestampMetadataInterceptor enabled?

@315157973 315157973 Jan 4, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unless we can ensure that the AppendBrokerTimestampMetadataInterceptor cannot be disabled, this bug may always occur

@codelipenghui codelipenghui Jan 5, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, I think we can either make the AppendBrokerTimestampMetadataInterceptor enabled by default or remove the AppendBrokerTimestampMetadataInterceptor to always apply the broker timestamp.

Without this change, the broker timestamp will be enabled without AppendBrokerTimestampMetadataInterceptor. Essentially, we delete the functionality of the AppendBrokerTimestampMetadataInterceptor.

@315157973 315157973 Jan 8, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I opened a discuss in dev thread https://lists.apache.org/thread/0vtjt436xvlg0fr4ogchkjdvjcp56mvz


int brokerMetaSize = brokerEntryMetadata.getSerializedSize();
ByteBuf brokerMeta =
Expand Down
22 changes: 22 additions & 0 deletions pulsar-common/src/test/java/org/apache/pulsar/common/protocol/CommandUtilsTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@

import static org.apache.pulsar.common.protocol.Commands.serializeMetadataAndPayload;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;

import io.netty.buffer.ByteBuf;
Expand Down Expand Up @@ -151,6 +153,26 @@ public void testAddBrokerEntryMetadata() throws Exception {
assertTrue(new String(content, StandardCharsets.UTF_8).endsWith(data));
}

@Test
public void testAlwaysSetBrokerTimestamp() {
int MOCK_BATCH_SIZE = 10;
String data = "test-message";
ByteBuf byteBuf = PulsarByteBufAllocator.DEFAULT.buffer(data.length(), data.length());
byteBuf.writeBytes(data.getBytes(StandardCharsets.UTF_8));
ByteBuf dataWithBrokerEntryMetadata =
Commands.addBrokerEntryMetadata(byteBuf, new HashSet<>(), MOCK_BATCH_SIZE);
BrokerEntryMetadata entryMetadata = Commands.parseBrokerEntryMetadataIfExist(dataWithBrokerEntryMetadata);
assertNotNull(entryMetadata);
assertTrue(entryMetadata.hasBrokerTimestamp());
long timestamp = entryMetadata.getBrokerTimestamp();
// add again, the timestamp should change
dataWithBrokerEntryMetadata =
Commands.addBrokerEntryMetadata(byteBuf, new HashSet<>(), MOCK_BATCH_SIZE);
entryMetadata = Commands.parseBrokerEntryMetadataIfExist(dataWithBrokerEntryMetadata);
assertNotNull(entryMetadata);
assertNotEquals(timestamp, entryMetadata.getBrokerTimestamp());
}

@Test
public void testSkipBrokerEntryMetadata() throws Exception {
String data = "test-message";
Expand Down