AW: When restarting an IoTDB with pre-existing pipeline subscriptions, the system doesn't correctly startup

[Christofer Dutz]

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 <christofer.d...@c-ware.de>
Datum: Dienstag, 13. August 2024 um 12:42
An: dev@iotdb.apache.org <dev@iotdb.apache.org>
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