Ok … using a full-text search and by setting breakpoints in every occasion that
could produce the log message I narrowed it down to this code:
The first one is logged by this code:
public void bindPrefetchingQueue(final SubscriptionConnectorSubtask subtask) {
final String consumerGroupId = subtask.getConsumerGroupId();
final SubscriptionBroker broker =
consumerGroupIdToSubscriptionBroker.get(consumerGroupId);
if (Objects.isNull(broker)) {
LOGGER.warn(
"Subscription: broker bound to consumer group [{}] does not exist",
consumerGroupId);
return;
}
broker.bindPrefetchingQueue(subtask.getTopicName(),
subtask.getInputPendingQueue());
}
After that all seem to be coming from here:
public void executePrefetch(final String consumerGroupId, final String
topicName) {
final SubscriptionBroker broker =
consumerGroupIdToSubscriptionBroker.get(consumerGroupId);
if (Objects.isNull(broker)) {
LOGGER.warn(
"Subscription: broker bound to consumer group [{}] does not exist",
consumerGroupId);
return;
}
broker.executePrefetch(topicName);
}
So, broker is always null and the method always directly returns … in
PipeSubtask:
@Override
public Boolean call() throws Exception {
boolean hasAtLeastOneEventProcessed = false;
try {
// If the scheduler allows to schedule, then try to consume an event
while (subtaskScheduler.schedule()) {
// If the event is consumed successfully, then continue to consume the
next event
// otherwise, stop consuming
if (!executeOnce()) {
break;
}
hasAtLeastOneEventProcessed = true;
}
} finally {
// Reset the scheduler to make sure that the scheduler can schedule again
subtaskScheduler.reset();
}
return hasAtLeastOneEventProcessed;
}
It's always treated as successful operation.
Not sure it should be this way.
Chris
Von: Christofer Dutz <[email protected]>
Datum: Dienstag, 13. August 2024 um 12:42
An: [email protected] <[email protected]>
Betreff: When restarting an IoTDB with pre-existing pipeline subscriptions, the
system doesn't correctly startup
Hi all,
I am using my Home-Automation experiment in parallel to my work in order to
learn and play with the cool features of IoTDB.
Here I have built a little Application, that uses PLC4X to subscribe to my KNX
network and pump all data into an IoTDB instance running embedded.
The thing I’m currently playing around with are the pipelines and the
subscription client.
So, when I start the system for the first time all is good … however as soon as
I restart the system, IoTDB floods my logger with so many:
14015 [pool-60-IoTDB-Subscription-Executor-Pool-5] WARN
o.a.i.d.s.a.SubscriptionBrokerAgent - Subscription: broker bound to consumer
group [weatherServiceGroup] does not exist
Messages, that I simply cannot spot the initial problem in my log.
Is there anything special that needs to be done when starting a system with
pre-existing subscriptions? Is it also possible to define them as “transient”?
I don’t really need IoTDB to buffer results till my application comes back up
(All I want is the changed values whenever something changes).
Chris
