Skip to content

Commit ecdf9e5

Browse files
aaron-aiaaron-ai
committed
Initial implementation
1 parent 447c3be commit ecdf9e5

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

50 files changed

+3936
-158
lines changed

‎.github/workflows/java_coverage.yml‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,8 @@ jobs:
1919
java-version: "11"
2020
distribution: "adopt"
2121
- name: Generate coverage report
22-
run: mvn test --file ./java/pom.xml
22+
working-directory: ./java
23+
run: mvn test --file ./pom.xml
2324
- name: Test summary
2425
uses: test-summary/action@v1
2526
with:

‎java/client-apis/src/main/java/org/apache/rocketmq/client/apis/SessionCredentials.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,26 +19,39 @@
1919

2020
import static com.google.common.base.Preconditions.checkNotNull;
2121

22+
import java.time.Duration;
2223
import java.util.Optional;
2324

2425
/**
2526
* Session credentials used in service authentications.
2627
*/
2728
public class SessionCredentials {
29+
private static final Duration EXPIRATION_BUFFER_TIME = Duration.ofSeconds(1);
2830
private final String accessKey;
2931
private final String accessSecret;
3032
private final String securityToken;
33+
private final long expiredTimestampMillis;
34+
35+
public SessionCredentials(String accessKey, String accessSecret, String securityToken,
36+
long expiredTimestampMillis) {
37+
this.accessKey = checkNotNull(accessKey, "accessKey should not be null");
38+
this.accessSecret = checkNotNull(accessSecret, "accessSecret should not be null");
39+
this.securityToken = checkNotNull(securityToken, "securityToken should not be null");
40+
this.expiredTimestampMillis = expiredTimestampMillis;
41+
}
3142

3243
public SessionCredentials(String accessKey, String accessSecret, String securityToken) {
3344
this.accessKey = checkNotNull(accessKey, "accessKey should not be null");
3445
this.accessSecret = checkNotNull(accessSecret, "accessSecret should not be null");
3546
this.securityToken = checkNotNull(securityToken, "securityToken should not be null");
47+
this.expiredTimestampMillis = Long.MAX_VALUE;
3648
}
3749

3850
public SessionCredentials(String accessKey, String accessSecret) {
3951
this.accessKey = checkNotNull(accessKey, "accessKey should not be null");
4052
this.accessSecret = checkNotNull(accessSecret, "accessSecret should not be null");
4153
this.securityToken = null;
54+
this.expiredTimestampMillis = Long.MAX_VALUE;
4255
}
4356

4457
public String getAccessKey() {
@@ -52,4 +65,8 @@ public String getAccessSecret() {
5265
public Optional<String> tryGetSecurityToken() {
5366
return null == securityToken ? Optional.empty() : Optional.of(securityToken);
5467
}
68+
69+
public boolean expiredSoon() {
70+
return System.currentTimeMillis() + EXPIRATION_BUFFER_TIME.toMillis() > expiredTimestampMillis;
71+
}
5572
}

‎java/client-apis/src/main/java/org/apache/rocketmq/client/apis/consumer/TopicMessageQueueChangeListener.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.rocketmq.client.apis.consumer;
19+
20+
import java.util.Set;
21+
import org.apache.rocketmq.client.apis.message.MessageQueue;
22+
23+
public interface TopicMessageQueueChangeListener {
24+
/**
25+
* This method will be invoked in the condition of queue numbers changed, These scenarios occur when the topic is
26+
* expanded or shrunk.
27+
*
28+
* @param topic the topic to listen.
29+
* @param messageQueues latest message queues of the topic.
30+
*/
31+
void onChanged(String topic, Set<MessageQueue> messageQueues);
32+
}

‎java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageQueue.java‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.rocketmq.client.apis.message;
19+
20+
public interface MessageQueue {
21+
/**
22+
* Topic of the current message queue.
23+
*/
24+
String getTopic();
25+
}

‎java/client-apis/src/main/java/org/apache/rocketmq/client/apis/producer/ProducerBuilder.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ public interface ProducerBuilder {
4848
* ArrayList<String> topicList = new ArrayList<>();
4949
* topicList.add("topicA");
5050
* topicList.add("topicB");
51-
* producerBuilder.setTopics(topicList);
51+
* producerBuilder.setTopics(topicList.toArray(new String[0]));
5252
* }</pre>
5353
*
5454
* @param topics topics to send/prepare.

‎java/client/src/main/java/org/apache/rocketmq/client/java/exception/StatusChecker.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@
1818
package org.apache.rocketmq.client.java.exception;
1919

2020
import apache.rocketmq.v2.Code;
21+
import apache.rocketmq.v2.GetOffsetRequest;
22+
import apache.rocketmq.v2.PullMessageRequest;
2123
import apache.rocketmq.v2.ReceiveMessageRequest;
2224
import apache.rocketmq.v2.Status;
2325
import org.apache.rocketmq.client.apis.ClientException;
@@ -61,6 +63,11 @@ public static void check(Status status, RpcFuture<?, ?> future) throws ClientExc
6163
case CLIENT_ID_REQUIRED:
6264
case ILLEGAL_POLLING_TIME:
6365
throw new BadRequestException(codeNumber, requestId, statusMessage);
66+
case ILLEGAL_OFFSET:
67+
if (future.getRequest() instanceof PullMessageRequest) {
68+
return;
69+
}
70+
// fall through on purpose.
6471
case UNAUTHORIZED:
6572
throw new UnauthorizedException(codeNumber, requestId, statusMessage);
6673
case PAYMENT_REQUIRED:
@@ -71,11 +78,19 @@ public static void check(Status status, RpcFuture<?, ?> future) throws ClientExc
7178
if (future.getRequest() instanceof ReceiveMessageRequest) {
7279
return;
7380
}
81+
if (future.getRequest() instanceof PullMessageRequest) {
82+
return;
83+
}
7484
// fall through on purpose.
7585
case NOT_FOUND:
7686
case TOPIC_NOT_FOUND:
7787
case CONSUMER_GROUP_NOT_FOUND:
7888
throw new NotFoundException(codeNumber, requestId, statusMessage);
89+
case OFFSET_NOT_FOUND:
90+
if (future.getRequest() instanceof GetOffsetRequest) {
91+
return;
92+
}
93+
// fall through on purpose.
7994
case PAYLOAD_TOO_LARGE:
8095
case MESSAGE_BODY_TOO_LARGE:
8196
throw new PayloadTooLargeException(codeNumber, requestId, statusMessage);

‎java/client/src/main/java/org/apache/rocketmq/client/java/hook/MessageHookPoints.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,10 @@ public enum MessageHookPoints {
2626
* The hook point of message reception.
2727
*/
2828
RECEIVE,
29+
/**
30+
* The hook point of message pulling.
31+
*/
32+
PULL,
2933
/**
3034
* The hook point of message consumption.
3135
*/

‎java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -588,7 +588,7 @@ public void onFailure(Throwable t) {
588588
public void doStats() {
589589
}
590590

591-
private ListenableFuture<TopicRouteData> fetchTopicRoute(final String topic) {
591+
protected ListenableFuture<TopicRouteData> fetchTopicRoute(final String topic) {
592592
final ListenableFuture<TopicRouteData> future0 = fetchTopicRoute0(topic);
593593
final ListenableFuture<TopicRouteData> future = Futures.transformAsync(future0,
594594
topicRouteData -> onTopicRouteDataFetched(topic, topicRouteData), MoreExecutors.directExecutor());

‎java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManager.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,19 +25,27 @@
2525
import apache.rocketmq.v2.EndTransactionResponse;
2626
import apache.rocketmq.v2.ForwardMessageToDeadLetterQueueRequest;
2727
import apache.rocketmq.v2.ForwardMessageToDeadLetterQueueResponse;
28+
import apache.rocketmq.v2.GetOffsetRequest;
29+
import apache.rocketmq.v2.GetOffsetResponse;
2830
import apache.rocketmq.v2.HeartbeatRequest;
2931
import apache.rocketmq.v2.HeartbeatResponse;
3032
import apache.rocketmq.v2.NotifyClientTerminationRequest;
3133
import apache.rocketmq.v2.NotifyClientTerminationResponse;
34+
import apache.rocketmq.v2.PullMessageRequest;
35+
import apache.rocketmq.v2.PullMessageResponse;
3236
import apache.rocketmq.v2.QueryAssignmentRequest;
3337
import apache.rocketmq.v2.QueryAssignmentResponse;
38+
import apache.rocketmq.v2.QueryOffsetRequest;
39+
import apache.rocketmq.v2.QueryOffsetResponse;
3440
import apache.rocketmq.v2.QueryRouteRequest;
3541
import apache.rocketmq.v2.QueryRouteResponse;
3642
import apache.rocketmq.v2.ReceiveMessageRequest;
3743
import apache.rocketmq.v2.ReceiveMessageResponse;
3844
import apache.rocketmq.v2.SendMessageRequest;
3945
import apache.rocketmq.v2.SendMessageResponse;
4046
import apache.rocketmq.v2.TelemetryCommand;
47+
import apache.rocketmq.v2.UpdateOffsetRequest;
48+
import apache.rocketmq.v2.UpdateOffsetResponse;
4149
import com.google.common.util.concurrent.AbstractIdleService;
4250
import io.grpc.stub.StreamObserver;
4351
import java.time.Duration;
@@ -147,6 +155,18 @@ public abstract RpcFuture<AckMessageRequest, AckMessageResponse> ackMessage(Endp
147155
ForwardMessageToDeadLetterQueueResponse> forwardMessageToDeadLetterQueue(Endpoints endpoints,
148156
ForwardMessageToDeadLetterQueueRequest request, Duration duration);
149157

158+
public abstract RpcFuture<PullMessageRequest, List<PullMessageResponse>> pullMessage(Endpoints endpoints,
159+
PullMessageRequest request, Duration duration);
160+
161+
public abstract RpcFuture<UpdateOffsetRequest, UpdateOffsetResponse> updateOffset(Endpoints endpoints,
162+
UpdateOffsetRequest request, Duration duration);
163+
164+
public abstract RpcFuture<GetOffsetRequest, GetOffsetResponse> getOffset(Endpoints endpoints,
165+
GetOffsetRequest request, Duration duration);
166+
167+
public abstract RpcFuture<QueryOffsetRequest, QueryOffsetResponse> queryOffset(Endpoints endpoints,
168+
QueryOffsetRequest request, Duration duration);
169+
150170
/**
151171
* Submit transaction resolution asynchronously, the method ensures no throwable.
152172
*

‎java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManagerImpl.java‎

Lines changed: 71 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -25,19 +25,27 @@
2525
import apache.rocketmq.v2.EndTransactionResponse;
2626
import apache.rocketmq.v2.ForwardMessageToDeadLetterQueueRequest;
2727
import apache.rocketmq.v2.ForwardMessageToDeadLetterQueueResponse;
28+
import apache.rocketmq.v2.GetOffsetRequest;
29+
import apache.rocketmq.v2.GetOffsetResponse;
2830
import apache.rocketmq.v2.HeartbeatRequest;
2931
import apache.rocketmq.v2.HeartbeatResponse;
3032
import apache.rocketmq.v2.NotifyClientTerminationRequest;
3133
import apache.rocketmq.v2.NotifyClientTerminationResponse;
34+
import apache.rocketmq.v2.PullMessageRequest;
35+
import apache.rocketmq.v2.PullMessageResponse;
3236
import apache.rocketmq.v2.QueryAssignmentRequest;
3337
import apache.rocketmq.v2.QueryAssignmentResponse;
38+
import apache.rocketmq.v2.QueryOffsetRequest;
39+
import apache.rocketmq.v2.QueryOffsetResponse;
3440
import apache.rocketmq.v2.QueryRouteRequest;
3541
import apache.rocketmq.v2.QueryRouteResponse;
3642
import apache.rocketmq.v2.ReceiveMessageRequest;
3743
import apache.rocketmq.v2.ReceiveMessageResponse;
3844
import apache.rocketmq.v2.SendMessageRequest;
3945
import apache.rocketmq.v2.SendMessageResponse;
4046
import apache.rocketmq.v2.TelemetryCommand;
47+
import apache.rocketmq.v2.UpdateOffsetRequest;
48+
import apache.rocketmq.v2.UpdateOffsetResponse;
4149
import com.google.common.util.concurrent.ListenableFuture;
4250
import com.google.errorprone.annotations.concurrent.GuardedBy;
4351
import io.grpc.Metadata;
@@ -283,8 +291,7 @@ public RpcFuture<AckMessageRequest, AckMessageResponse> ackMessage(Endpoints end
283291

284292
@Override
285293
public RpcFuture<ChangeInvisibleDurationRequest, ChangeInvisibleDurationResponse>
286-
changeInvisibleDuration(Endpoints endpoints, ChangeInvisibleDurationRequest request,
287-
Duration duration) {
294+
changeInvisibleDuration(Endpoints endpoints, ChangeInvisibleDurationRequest request, Duration duration) {
288295
try {
289296
final Metadata metadata = client.sign();
290297
final Context context = new Context(endpoints, metadata);
@@ -313,6 +320,66 @@ public RpcFuture<AckMessageRequest, AckMessageResponse> ackMessage(Endpoints end
313320
}
314321
}
315322

323+
@Override
324+
public RpcFuture<PullMessageRequest, List<PullMessageResponse>> pullMessage(Endpoints endpoints,
325+
PullMessageRequest request, Duration duration) {
326+
try {
327+
final Metadata metadata = client.sign();
328+
final Context context = new Context(endpoints, metadata);
329+
final RpcClient rpcClient = getRpcClient(endpoints);
330+
final ListenableFuture<List<PullMessageResponse>> future =
331+
rpcClient.pullMessage(metadata, request, asyncWorker, duration);
332+
return new RpcFuture<>(context, request, future);
333+
} catch (Throwable t) {
334+
return new RpcFuture<>(t);
335+
}
336+
}
337+
338+
@Override
339+
public RpcFuture<UpdateOffsetRequest, UpdateOffsetResponse> updateOffset(Endpoints endpoints,
340+
UpdateOffsetRequest request, Duration duration) {
341+
try {
342+
final Metadata metadata = client.sign();
343+
final Context context = new Context(endpoints, metadata);
344+
final RpcClient rpcClient = getRpcClient(endpoints);
345+
final ListenableFuture<UpdateOffsetResponse> future =
346+
rpcClient.updateOffset(metadata, request, asyncWorker, duration);
347+
return new RpcFuture<>(context, request, future);
348+
} catch (Throwable t) {
349+
return new RpcFuture<>(t);
350+
}
351+
}
352+
353+
@Override
354+
public RpcFuture<GetOffsetRequest, GetOffsetResponse> getOffset(Endpoints endpoints, GetOffsetRequest request,
355+
Duration duration) {
356+
try {
357+
final Metadata metadata = client.sign();
358+
final Context context = new Context(endpoints, metadata);
359+
final RpcClient rpcClient = getRpcClient(endpoints);
360+
final ListenableFuture<GetOffsetResponse> future =
361+
rpcClient.getOffset(metadata, request, asyncWorker, duration);
362+
return new RpcFuture<>(context, request, future);
363+
} catch (Throwable t) {
364+
return new RpcFuture<>(t);
365+
}
366+
}
367+
368+
@Override
369+
public RpcFuture<QueryOffsetRequest, QueryOffsetResponse> queryOffset(Endpoints endpoints,
370+
QueryOffsetRequest request, Duration duration) {
371+
try {
372+
final Metadata metadata = client.sign();
373+
final Context context = new Context(endpoints, metadata);
374+
final RpcClient rpcClient = getRpcClient(endpoints);
375+
final ListenableFuture<QueryOffsetResponse> future =
376+
rpcClient.queryOffset(metadata, request, asyncWorker, duration);
377+
return new RpcFuture<>(context, request, future);
378+
} catch (Throwable t) {
379+
return new RpcFuture<>(t);
380+
}
381+
}
382+
316383
@Override
317384
public RpcFuture<EndTransactionRequest, EndTransactionResponse> endTransaction(Endpoints endpoints,
318385
EndTransactionRequest request, Duration duration) {
@@ -395,9 +462,9 @@ protected void startUp() {
395462
() -> {
396463
try {
397464
log.info("Start to log statistics, clientVersion={}, clientWrapperVersion={}, "
398-
+ "clientEndpoints={}, os description=[{}], java description=[{}], clientId={}",
465+
+ "clientEndpoints={}, os description=[{}], java environment=[{}], clientId={}",
399466
MetadataUtils.getVersion(), MetadataUtils.getWrapperVersion(), client.getEndpoints(),
400-
Utilities.getOsDescription(), Utilities.getJavaDescription(), clientId);
467+
Utilities.getOsDescription(), Utilities.getJavaEnvironmentSummary(), clientId);
401468
client.doStats();
402469
} catch (Throwable t) {
403470
log.error("Exception raised during statistics logging, clientId={}", clientId, t);

0 commit comments

Comments
 (0)