forked from apache/pulsar
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[improve][client] PIP-229: Add a common interface to get fields of th…
…e MessageIdData Master issue: apache#18950 ### Motivation We need a common interface to get fields of the MessageIdData. After that, we won't need to assert a MessageId implementation is an instance of a specific class. And we can pass our customized MessageId implementation to APIs like `acknowledge` and `seek`. ### Modifications - Add `MessageIdAdv` to get fields of `MessageIdData`, make all MessageId implementations inherit it (except `MultiMessageIdImpl`). - Deprecate `BatchMessageAcker` by adding the ACK bit set field and the `PreviousMessageAcknowledger` interface to `BatchMessageIdImpl`. - Deprecate `TopicMessageIdImpl#getInnerMessageId` by passing the `TopicMessageIdImpl` directly. - Remove `instanceof BatchMessageIdImpl` checks in `pulsar-client` module by casting to `MessageIdAdv`.
- Loading branch information
1 parent
fcecca4
commit ee35d7d
Showing
31 changed files
with
519 additions
and
613 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
122 changes: 122 additions & 0 deletions
122
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/MessageIdAdv.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,122 @@ | ||
| /* | ||
| * 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.pulsar.client.api; | ||
|
|
||
| import java.util.BitSet; | ||
|
|
||
| /** | ||
| * The {@link MessageId} interface provided for advanced users. | ||
| * <p> | ||
| * All built-in MessageId implementations should be able to be cast to MessageIdAdv. | ||
| * </p> | ||
| */ | ||
| public interface MessageIdAdv extends MessageId { | ||
|
|
||
| /** | ||
| * Get the ledger ID. | ||
| * | ||
| * @return the ledger ID | ||
| */ | ||
| long getLedgerId(); | ||
|
|
||
| /** | ||
| * Get the entry ID. | ||
| * | ||
| * @return the entry ID | ||
| */ | ||
| long getEntryId(); | ||
|
|
||
| /** | ||
| * Get the partition index. | ||
| * | ||
| * @return -1 if the message is from a non-partitioned topic, otherwise the non-negative partition index | ||
| */ | ||
| default int getPartitionIndex() { | ||
| return -1; | ||
| } | ||
|
|
||
| /** | ||
| * Get the batch index. | ||
| * | ||
| * @return -1 if the message is not in a batch | ||
| */ | ||
| default int getBatchIndex() { | ||
| return -1; | ||
| } | ||
|
|
||
| /** | ||
| * Get the batch size. | ||
| * | ||
| * @return 0 if the message is not in a batch | ||
| */ | ||
| default int getBatchSize() { | ||
| return 0; | ||
| } | ||
|
|
||
| /** | ||
| * Get the BitSet that indicates which messages in the batch. | ||
| * | ||
| * @implNote The message IDs of a batch should share a BitSet. For example, given 3 messages in the same batch whose | ||
| * size is 3, all message IDs of them should return "111" (i.e. a BitSet whose size is 3 and all bits are 1). If the | ||
| * 1st message has been acknowledged, the returned BitSet should become "011" (i.e. the 1st bit become 0). | ||
| * | ||
| * @return null if the message is a non-batched message | ||
| */ | ||
| default BitSet getAckSet() { | ||
| return null; | ||
| } | ||
|
|
||
| /** | ||
| * Get the message ID of the first chunk if the current message ID represents the position of a chunked message. | ||
| * | ||
| * @implNote A chunked message is distributed across different BookKeeper entries. The message ID of a chunked | ||
| * message is composed of two message IDs that represent positions of the first and the last chunk. The message ID | ||
| * itself represents the position of the last chunk. | ||
| * | ||
| * @return null if the message is not a chunked message | ||
| */ | ||
| default MessageIdAdv getFirstChunkMessageId() { | ||
| return null; | ||
| } | ||
|
|
||
| /** | ||
| * The default implementation of {@link Comparable#compareTo(Object)}. | ||
| */ | ||
| default int compareTo(MessageId o) { | ||
| if (!(o instanceof MessageIdAdv)) { | ||
| throw new UnsupportedOperationException("Unknown MessageId type: " | ||
| + ((o != null) ? o.getClass().getName() : "null")); | ||
| } | ||
| final MessageIdAdv other = (MessageIdAdv) o; | ||
| int result = Long.compare(this.getLedgerId(), other.getLedgerId()); | ||
| if (result != 0) { | ||
| return result; | ||
| } | ||
| result = Long.compare(this.getEntryId(), other.getEntryId()); | ||
| if (result != 0) { | ||
| return result; | ||
| } | ||
| // TODO: Correct the following compare logics, see https://github.com/apache/pulsar/pull/18981 | ||
| result = Integer.compare(this.getPartitionIndex(), other.getPartitionIndex()); | ||
| if (result != 0) { | ||
| return result; | ||
| } | ||
| return Integer.compare(this.getBatchIndex(), other.getBatchIndex()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -20,6 +20,7 @@ | |
|
|
||
| import java.util.BitSet; | ||
|
|
||
| @Deprecated | ||
| public class BatchMessageAcker { | ||
|
|
||
| private BatchMessageAcker() { | ||
|
|
||
50 changes: 0 additions & 50 deletions
50
pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageAckerDisabled.java
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.