Slack digest for #general - 2020-08-16

[Apache Pulsar Slack <ap...@gmail.com>]

2020-08-15 10:38:03 UTC - Christophe Bornet: For the API, I'm thinking of a way to register something like a gRPC CallStreamObserver ( <https://grpc.github.io/grpc-java/javadoc/io/grpc/stub/CallStreamObserver.html|https://grpc.github.io/grpc-java/javadoc/io/grpc/stub/CallStreamObserver.html>) or a reactive streams Subscription (<https://www.reactive-streams.org/reactive-streams-1.0.0-javadoc/org/reactivestreams/Subscription.html#request-long-|https://www.reactive-streams.org/reactive-streams-1.0.0-javadoc/org/reactivestreams/Subscription.html#request-long->) which have a request(n) method to signal how much items the producer can push. At startup the request would be called with the size of the queue. And then request(1) is called each time an item is removed from the queue.
----
2020-08-15 12:29:58 UTC - Fernando: That article confirms my understanding that  messages shouldn’t be removed from the subscription’s backlog if unacknowledged. The problem is that if I look at the topic stats, the subscription is just gone :s
+1 : Julius S
----
2020-08-15 12:39:09 UTC - Fernando: It looks similar to this bug <https://github.com/apache/pulsar/issues/5579|https://github.com/apache/pulsar/issues/5579> however my namespace doesn’t have a TTL. Instead it has infinite retention, which makes it weirder
----
2020-08-15 15:43:52 UTC - Joshua Decosta: ```    public static ByteBuf newConnect(String authMethodName, AuthData authData, int protocolVersion, String libVersion,
                                     String targetBroker, String originalPrincipal, AuthData originalAuthData,
                                     String originalAuthMethod) {
        CommandConnect.Builder connectBuilder = CommandConnect.newBuilder();
        connectBuilder.setClientVersion(libVersion != null ? libVersion : "Pulsar Client");
        connectBuilder.setAuthMethodName(authMethodName);

        if (targetBroker != null) {
            // When connecting through a proxy, we need to specify which broker do we want to be proxied through
            connectBuilder.setProxyToBrokerUrl(targetBroker);
        }

        if (authData != null) {
            connectBuilder.setAuthData(ByteString.copyFrom(authData.getBytes()));
        }

        if (originalPrincipal != null) {
            connectBuilder.setOriginalPrincipal(originalPrincipal);
        }

        if (originalAuthData != null) {
            connectBuilder.setOriginalAuthData(new String(originalAuthData.getBytes(), UTF_8));
        }

        if (originalAuthMethod != null) {
            connectBuilder.setOriginalAuthMethod(originalAuthMethod);
        }
        connectBuilder.setProtocolVersion(protocolVersion);
        connectBuilder.setFeatureFlags(getFeatureFlags());
        CommandConnect connect = connectBuilder.build();
        ByteBuf res = serializeWithSize(BaseCommand.newBuilder().setType(Type.CONNECT).setConnect(connect));
        connect.recycle();
        connectBuilder.recycle();
        return res;
    }```
After debugging for a bit I keep hitting this method in `Commands.java`  and I see it's setting the original prinicple with the originalprinciple auth data. Why is this happening? Am I supposed to be looking for the original principle auth data my `AuthorizationProvider` methods? I still can't figure out why the data isn't being passed.
----
2020-08-15 15:56:59 UTC - Joshua Decosta: I'm not sure where I should be looking at this point.
----
2020-08-15 16:03:34 UTC - Joshua Decosta: This method in `ProxyConnection` gets hit often:

```    private void completeConnect() {
        <http://LOG.info|LOG.info>("[{}] complete connection, init proxy handler. authenticated with {} role {}, hasProxyToBrokerUrl: {}",
            remoteAddress, authMethod, clientAuthRole, hasProxyToBrokerUrl);
        if (hasProxyToBrokerUrl) {
            // Client already knows which broker to connect. Let's open a
            // connection there and just pass bytes in both directions
            state = State.ProxyConnectionToBroker;
            directProxyHandler = new DirectProxyHandler(service, this, proxyToBrokerUrl,
                protocolVersionToAdvertise, sslHandlerSupplier);
            cancelKeepAliveTask();
        } else {
            // Client is doing a lookup, we can consider the handshake complete
            // and we'll take care of just topics and
            // partitions metadata lookups
            state = State.ProxyLookupRequests;
            lookupProxyHandler = new LookupProxyHandler(service, this);
            ctx.writeAndFlush(Commands.newConnected(protocolVersionToAdvertise));
        }
    }```
That else statement gets hit almost everytime.
----