BarrageSubscription NPE when empty update is received after initial subscription

[abaranec]

Description

If an empty update is received on a Barrage subscription after the initial subscription has been completed, you get an NPE

Steps to reproduce

1. Create a DH instance with the following script code

import io.deephaven.engine.util.SortedBy
qq = timeTable("PT1S")
qqWatch = SortedBy.sortedLastBy(qq, "Timestamp")

2. In another DH instance, try to subscribe to that table with barrage

import io.grpc.netty.NettyChannelBuilder
import io.deephaven.proto.DeephavenChannelImpl
import io.deephaven.client.impl.BarrageSession
import io.deephaven.client.impl.SessionImpl
import io.deephaven.client.impl.SessionImplConfig
import java.util.concurrent.Executors
import org.apache.arrow.memory.RootAllocator
import io.deephaven.qst.table.TicketTable
import io.deephaven.server.uri.BarrageTableResolver

managedChannel = NettyChannelBuilder.forTarget("dh://REMOTE_HOST").build()
dhChannel = new DeephavenChannelImpl(managedChannel)
session = SessionImpl.create(SessionImplConfig.builder()
    .executor(Executors.newScheduledThreadPool(4))
//    .authenticationTypeAndValue(AUTH_DETAILS)
    .channel(dhChannel)
    .build())
barrage = BarrageSession.of(session, new RootAllocator(), managedChannel)

sub = barrage.subscribe(TicketTable.of("s/qqWatch"), BarrageTableResolver.SUB_OPTIONS)
table = sub.entireTable()

Expected results

You get a functional subscription

Actual results

You get an NPE on the first empty table update

Versions

• Deephaven: 0.28.0

[rcaudy]

We're invoking a listener on the wrong thread, potentially without the mutual-exclusion or happens-before boundary we expected. It's racing with the update processing that needs to actually happen for the completion condition to be properly detected. It might even be racing with another notification that is supposed to trigger completion; that is, we might trigger completion early, from the wrong thread, in the case where we get a message with only removes.

Let's start by reverting the original change and making a deeper one.

1. We need to be delivering this completion notification after processing the barrage message from the update graph root. 2. The "extra" completion check should only happen on the sole message delivered for the initial snapshot of an empty table.
2. We seem to have some sloppiness/raciness with error handling. I don't see how an error will propagate from BarrageSubscriptionImpl or BarrageSnapshotImpl back to a blocked caller, for instance.