[Bug] Cannot invoke "apache.rocketmq.v2.Settings.getSubscription()" because "settings" is null. NullPointerException.
- Dominant language
- Java
- Stars
- 22.6k
- Forks
- 12k
- Avg merge
- 2d 20h
- Merged PRs (30d)
- 26
Description
### Before Creating the Bug Report
- [X] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq/discussions).
- [X] I have searched the [GitHub Issues](https://github.com/apache/rocketmq/issues) and [GitHub Discussions](https://github.com/apache/rocketmq/discussions) of this repository and believe that this is not a duplicate.
- [X] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.
### Runtime platform environment
Macos Node.js 18.17.1
### RocketMQ version
RocketMQ 5.3.0
### JDK Version
openjdk 17.0.11 2024-04-16
OpenJDK Runtime Environment Homebrew (build 17.0.11+0)
OpenJDK 64-Bit Server VM Homebrew (build 17.0.11+0, mixed mode, sharing)
### Describe the Bug
I got an inexplicable error. I just started and waited for messages, did nothing, and suddenly an error occurred. I caught it. I am not sure whether I need to ignore this error. It looks more like a bug or a problem that should not exist. It will interrupt my program.
`receive message error InternalErrorException: [request-id=undefined, response-code=50001] Cannot invoke "apache.rocketmq.v2.Settings.getSubscription()" because "settings" is null. NullPointerException. org.apache.rocketmq.proxy.grpc.v2.consumer.ReceiveMessageActivity.receiveMessage(ReceiveMessageActivity.java:63)
at Function.check (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/exception/StatusChecker.ts:81:15)
at SimpleConsumer.receiveMessage (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/consumer/Consumer.ts:78:19)
at processTicksAndRejections (node:internal/process/task_queues:95:5)
at SimpleConsumer.receive (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/consumer/SimpleConsumer.ts:122:12)
at AppController.listenForMessages (/Users/muyouzhi/Code/demo/nest-demo/src/app.controller.ts:107:24) {
code: 50001
}
/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/exception/StatusChecker.ts:81
throw new InternalErrorException(status.code, status.message, requestId);
^
InternalErrorException: [request-id=undefined, response-code=50001] Cannot invoke "apache.rocketmq.v2.Settings.getSubscription()" because "settings" is null. NullPointerException. org.apache.rocketmq.proxy.grpc.v2.consumer.ReceiveMessageActivity.receiveMessage(ReceiveMessageActivity.java:63)
at Function.check (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/exception/StatusChecker.ts:81:15)
at SimpleConsumer.receiveMessage (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/consumer/Consumer.ts:78:19)
at processTicksAndRejections (node:internal/process/task_queues:95:5)
at SimpleConsumer.receive (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/consumer/SimpleConsumer.ts:122:12)
at AppController.listenForMessages (/Users/muyouzhi/Code/demo/nest-demo/src/app.controller.ts:107:24)
`
### Steps to Reproduce
Start and run the following code in the Nest program, and then try to send a few messages. It should be normal at this time.
Then we wait patiently for a while. This error will not be triggered in a short time. It seems that there is a scheduled task that changes some status. In short, we have to wait for a long time.
```ts
import { Body, Controller, OnModuleInit, Post } from "@nestjs/common";
import { MessageOptions, Producer, SimpleConsumer } from "rocketmq-client-nodejs";
interface IProducerQuery {
topic: string;
tag: string;
keys?: string;
body: string;
group?: string;
num?: number;
}
@Controller()
export class AppController implements OnModuleInit {
private producer: Producer;
async onModuleInit() {
await this.producerInit();
await this.consumerInit();
}
async producerInit() {
const producer = new Producer({
endpoints: "127.0.0.1:8081"
});
this.producer = producer;
await this.producer.startup()
.catch((e) => {
console.error("producer startup error", e);
process.exit(1);
})
.then(() => {
console.log("producer startup success!");
});
}
async consumerInit() {
const task_to_pool_consumer = new SimpleConsumer({
consumerGroup: "task-to-pool",
endpoints: "127.0.0.1:8081",
subscriptions: new Map().set("task-processing", "task-to-pool")
});
await task_to_pool_consumer.startup()
.catch((e) => {
console.error("consumer startup error", e);
process.exit(1);
})
.then(() => {
console.log("task to pool consumer startup success!");
this.listenForMessages(task_to_pool_consumer, 5);
});
const kitchens = ["kitchen1", "kitchen2", "kitchen3", "kitchen4", "kitchen5"];
const kitchen_consumers: SimpleConsumer[] = [];
for (const kitchen of kitchens) {
const task_assign_consumer = new SimpleConsumer({
consumerGroup: "task-assign",
endpoints: "127.0.0.1:8081",
subscriptions: new Map().set("task-processing", kitchen)
});
kitchen_consumers.push(task_assign_consumer);
}
await Promise.all(kitchen_consumers.map((consumer: SimpleConsumer) => {
consumer.startup().catch((e) => {
console.error("consumer startup error", e);
process.exit(1);
}).then(() => {
console.log("kitchen consumer startup success!");
this.listenForMessages(consumer, 5);
});
}));
const task_to_db_consumer = new SimpleConsumer({
consumerGroup: "task-to-db",
endpoints: "127.0.0.1:8081",
subscriptions: new Map().set("task-write", "task-to-db")
});
await task_to_db_consumer.startup()
.catch((e) => {
console.error("consumer startup error", e);
process.exit(1);
})
.then(() => {
console.log("task to db consumer startup success!");
this.listenForMessages(task_to_db_consumer, 1);
});
const task_to_cache_consumer = new SimpleConsumer({
consumerGroup: "task-to-cache",
endpoints: "127.0.0.1:8081",
subscriptions: new Map().set("task-write", "task-to-cache")
});
await task_to_cache_consumer.startup()
.catch((e) => {
console.error("consumer startup error", e);
process.exit(1);
})
.then(() => {
console.log("task to cache consumer startup success!");
this.listenForMessages(task_to_cache_consumer, 1);
});
}
async listenForMessages(consumer: SimpleConsumer, number: number) {
while (true) {
const messages = await consumer.receive(number).catch((e) => {
console.error("receive message error", e);
throw e;
});
if (!messages || messages.length === 0) {
continue;
}
for (const message of messages) {
console.log(`Received message: ${message.body.toString()}`);
await consumer.ack(message).catch((e) => {
console.error("ack message error", e);
throw e;
}).then(() => {
console.log("ack message success!");
});
}
}
}
@Post("production")
async production(@Body() query: IProducerQuery) {
const options: MessageOptions = {
topic: query.topic,
tag: query.tag,
keys: [query.keys as string] || [Date.now().toString()],
body: Buffer.from(query.body)
};
query.group ? options.messageGroup = query.group : null;
if (query.num && query.num > 1) {
for (let i = 0; i < query.num; i++) {
options.keys = [Date.now().toString()];
options.body = Buffer.from(query.body + i);
await this.producer
.send(options)
.catch((e) => {
console.error("producer send message error", e);
})
.then((r) => {
console.log(`producer send message success! receipt -> ${r ? JSON.stringify(r) : "null"}`);
});
}
return;
}
await this.producer
.send(options)
.catch((e) => {
console.error("producer send message error", e);
})
.then((r) => {
console.log(`producer send message success! receipt -> ${r ? JSON.stringify(r) : "null"}`);
});
}
}
```
If nothing goes wrong, you should wait until the program has an error:
```
receive message error InternalErrorException: [request-id=undefined, response-code=50001] Cannot invoke "apache.rocketmq.v2.Settings.getSubscription()" because "settings" is null. NullPointerException. org.apache.rocketmq.proxy.grpc.v2.consumer.ReceiveMessageActivity.receiveMessage(ReceiveMessageActivity.java:63)
at Function.check (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/exception/StatusChecker.ts:81:15)
at SimpleConsumer.receiveMessage (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/consumer/Consumer.ts:78:19)
at processTicksAndRejections (node:internal/process/task_queues:95:5)
at SimpleConsumer.receive (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/consumer/SimpleConsumer.ts:122:12)
at AppController.listenForMessages (/Users/muyouzhi/Code/demo/nest-demo/src/app.controller.ts:107:24) {
code: 50001
}
/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/exception/StatusChecker.ts:81
throw new InternalErrorException(status.code, status.message, requestId);
^
InternalErrorException: [request-id=undefined, response-code=50001] Cannot invoke "apache.rocketmq.v2.Settings.getSubscription()" because "settings" is null. NullPointerException. org.apache.rocketmq.proxy.grpc.v2.consumer.ReceiveMessageActivity.receiveMessage(ReceiveMessageActivity.java:63)
at Function.check (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/exception/StatusChecker.ts:81:15)
at SimpleConsumer.receiveMessage (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/consumer/Consumer.ts:78:19)
at processTicksAndRejections (node:internal/process/task_queues:95:5)
at SimpleConsumer.receive (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/consumer/SimpleConsumer.ts:122:12)
at AppController.listenForMessages (/Users/muyouzhi/Code/demo/nest-demo/src/app.controller.ts:107:24)
```
This error comes from StatusChecker.check(status) in the receiveMessage method.
This error indicates that when calling RocketMQ to receive messages, Settings.getSubscription() returned null, causing a NullPointerException.
Why does the Subscription data in Settings suddenly disappear? Why does the NullPointerException exist?
### What Did You Expect to See?
Continuous, continuous and normal operation
### What Did You See Instead?
```
receive message error InternalErrorException: [request-id=undefined, response-code=50001] Cannot invoke "apache.rocketmq.v2.Settings.getSubscription()" because "settings" is null. NullPointerException. org.apache.rocketmq.proxy.grpc.v2.consumer.ReceiveMessageActivity.receiveMessage(ReceiveMessageActivity.java:63)
at Function.check (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/exception/StatusChecker.ts:81:15)
at SimpleConsumer.receiveMessage (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/consumer/Consumer.ts:78:19)
at processTicksAndRejections (node:internal/process/task_queues:95:5)
at SimpleConsumer.receive (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/consumer/SimpleConsumer.ts:122:12)
at AppController.listenForMessages (/Users/muyouzhi/Code/demo/nest-demo/src/app.controller.ts:107:24) {
code: 50001
}
/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/exception/StatusChecker.ts:81
throw new InternalErrorException(status.code, status.message, requestId);
^
InternalErrorException: [request-id=undefined, response-code=50001] Cannot invoke "apache.rocketmq.v2.Settings.getSubscription()" because "settings" is null. NullPointerException. org.apache.rocketmq.proxy.grpc.v2.consumer.ReceiveMessageActivity.receiveMessage(ReceiveMessageActivity.java:63)
at Function.check (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/exception/StatusChecker.ts:81:15)
at SimpleConsumer.receiveMessage (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/consumer/Consumer.ts:78:19)
at processTicksAndRejections (node:internal/process/task_queues:95:5)
at SimpleConsumer.receive (/Users/muyouzhi/Code/demo/nest-demo/node_modules/.pnpm/rocketmq-client-nodejs@1.0.0/node_modules/rocketmq-client-nodejs/src/consumer/SimpleConsumer.ts:122:12)
at AppController.listenForMessages (/Users/muyouzhi/Code/demo/nest-demo/src/app.controller.ts:107:24)
```
### Additional Context
_No response_
Contributor guide
Research direction
Start at org.apache.rocketmq.proxy.grpc.v2.consumer.ReceiveMessageActivity.receiveMessage(ReceiveMessageActivity.java:63) and trace how the request reaches StatusChecker.ts:81, Consumer.ts, and SimpleConsumer.ts. Reproduce the long-running receive scenario from the NestJS example and determine why the Settings value becomes null; done means continuous receiving no longer produces this NullPointerException.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, node.js, typescript
- Domain
- api, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 38/100