[fix][auth] Allow natural exit with token suppliers

src/Authentication.cc

@@ -26,8 +26,6 @@ static const std::string PARAM_TOKEN = "token";
 static const std::string PARAM_USERNAME = "username";
 static const std::string PARAM_PASSWORD = "password";
 
-void FinalizeTokenSupplierCallback(Napi::Env env, TokenSupplierCallback *cb, void *) { delete cb; }
-
 struct TokenSupplierProxyData {
   std::function<void(void)> callback;
   std::string token;
@@ -134,14 +132,16 @@ Authentication::Authentication(const Napi::CallbackInfo &info)
         }
 
         if (obj.Get(PARAM_TOKEN).IsFunction()) {
-          this->tokenSupplier = new TokenSupplierCallback();
+          this->tokenSupplier = std::make_shared<TokenSupplierCallback>();
+          std::shared_ptr<TokenSupplierCallback> tokenSupplier = this->tokenSupplier;
           Napi::ThreadSafeFunction callback = Napi::ThreadSafeFunction::New(
               obj.Env(), obj.Get(PARAM_TOKEN).As<Napi::Function>(), "Token Supplier Callback", 1, 1,
-              (void *)NULL, FinalizeTokenSupplierCallback, this->tokenSupplier);
+              [tokenSupplier](Napi::Env) { tokenSupplier->finalized = true; });
           this->tokenSupplier->callback = std::move(callback);
+          this->tokenSupplier->callback.Unref(env);
 
           this->cAuthentication =
-              pulsar_authentication_token_create_with_supplier(&TokenSupplier, this->tokenSupplier);
+              pulsar_authentication_token_create_with_supplier(&TokenSupplier, this->tokenSupplier.get());
           return;
         }
       }
@@ -178,7 +178,7 @@ Authentication::Authentication(const Napi::CallbackInfo &info)
 }
 
 Authentication::~Authentication() {
-  if (this->tokenSupplier != nullptr) {
+  if (this->tokenSupplier != nullptr && !this->tokenSupplier->finalized) {
     this->tokenSupplier->callback.Release();
   }
   if (this->cAuthentication != nullptr) {

src/Authentication.h

@@ -20,6 +20,7 @@
 #ifndef AUTH_H
 #define AUTH_H
 
+#include <memory>
 #include <napi.h>
 #include <pulsar/c/authentication.h>
 #include "TokenSupplier.h"
@@ -35,7 +36,7 @@ class Authentication : public Napi::ObjectWrap<Authentication> {
   static Napi::FunctionReference constructor;
   pulsar_authentication_t *cAuthentication;
 
-  TokenSupplierCallback *tokenSupplier;
+  std::shared_ptr<TokenSupplierCallback> tokenSupplier;
 };
 
 #endif

src/TokenSupplier.h

@@ -24,6 +24,7 @@
 
 struct TokenSupplierCallback {
   Napi::ThreadSafeFunction callback;
+  bool finalized = false;
 
   TokenSupplierCallback() {}
 };

tests/process_exit.test.js

@@ -0,0 +1,102 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+const path = require('path');
+const { spawn } = require('child_process');
+
+const scenariosDir = path.join(__dirname, 'process_exit');
+const runner = path.join(scenariosDir, 'runner.js');
+
+const expectNaturalExit = (script, args) => new Promise((resolve, reject) => {
+  const child = spawn(process.execPath, [runner, script, ...args], {
+    stdio: ['ignore', 'pipe', 'pipe', 'ipc'],
+  });
+  let stdout = '';
+  let stderr = '';
+  let spawnError;
+  let protocolFailure;
+  let timedOut;
+  let completed = false;
+  let disconnectedAfterCompletion = false;
+  let naturalExitTimer;
+
+  const watchdogTimer = setTimeout(() => {
+    timedOut = completed ? 'IPC disconnect' : 'scenario completion';
+    child.kill('SIGKILL');
+  }, 60000);
+
+  child.stdout.on('data', (data) => {
+    stdout += data.toString();
+  });
+  child.stderr.on('data', (data) => {
+    stderr += data.toString();
+  });
+  child.on('message', (message) => {
+    if (message && message.type === 'scenario-complete') {
+      completed = true;
+    }
+  });
+  child.on('disconnect', () => {
+    if (!completed) {
+      protocolFailure = 'IPC disconnected before scenario completion';
+      child.kill('SIGKILL');
+      return;
+    }
+
+    disconnectedAfterCompletion = true;
+    clearTimeout(watchdogTimer);
+    naturalExitTimer = setTimeout(() => {
+      timedOut = 'natural exit after scenario completion and IPC disconnect';
+      child.kill('SIGKILL');
+    }, 10000);
+  });
+  child.on('error', (error) => {
+    spawnError = error;
+    clearTimeout(watchdogTimer);
+    clearTimeout(naturalExitTimer);
+  });
+  child.on('close', (code, signal) => {
+    clearTimeout(watchdogTimer);
+    clearTimeout(naturalExitTimer);
+    if (timedOut || spawnError || protocolFailure || !completed
+      || !disconnectedAfterCompletion || code !== 0 || signal !== null) {
+      reject(new Error(`Child did not exit naturally: ${JSON.stringify({
+        timedOut,
+        spawnError: spawnError && spawnError.message,
+        protocolFailure,
+        completed,
+        disconnectedAfterCompletion,
+        code,
+        signal,
+        stdout,
+        stderr,
+      })}`));
+      return;
+    }
+    resolve();
+  });
+});
+
+describe('AuthenticationToken supplier process exit', () => {
+  const scenario = path.join(scenariosDir, 'token_supplier.js');
+
+  test.each(['sync', 'async'])('%s supplier exits after closing a client', async (supplier) => {
+    await expectNaturalExit(scenario, [supplier]);
+  }, 75000);
+});

tests/process_exit/runner.js

@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+const disconnect = () => {
+  if (process.connected) {
+    process.disconnect();
+  }
+};
+
+const fail = (error) => {
+  console.error(error);
+  process.exitCode = 1;
+  disconnect();
+};
+
+const runScenario = async () => {
+  if (typeof process.send !== 'function') {
+    throw new Error('Scenario runner requires an IPC channel');
+  }
+
+  // eslint-disable-next-line global-require, import/no-dynamic-require
+  const scenario = require(process.argv[2]);
+  if (typeof scenario !== 'function') {
+    throw new TypeError('Scenario module must export a function');
+  }
+  await scenario(...process.argv.slice(3));
+};
+
+runScenario().then(() => {
+  process.send({ type: 'scenario-complete' }, (error) => {
+    if (error) {
+      fail(error);
+      return;
+    }
+    disconnect();
+  });
+}).catch(fail);

tests/process_exit/token_supplier.js

@@ -0,0 +1,39 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+const Pulsar = require('../../index');
+
+const suppliers = {
+  sync: () => 'dummy-token',
+  async: async () => 'dummy-token',
+};
+
+module.exports = async (mode) => {
+  const supplier = suppliers[mode];
+  if (!supplier) {
+    throw new Error(`Unknown supplier: ${mode}`);
+  }
+
+  global.auth = new Pulsar.AuthenticationToken({ token: supplier });
+  const client = new Pulsar.Client({
+    serviceUrl: 'pulsar://localhost:6650',
+    authentication: global.auth,
+  });
+  await client.close();
+};