title: Custom plugin for librdkafka registering oauthbearer_token_refresh_cb not being called
body: <p>I'm trying to write custom plugin for <code>librdkafka</code> which will provide custom implementation of <code>oauthbearer_token_refresh_cb</code>.</p>
  <p>In our company we use kafka with custom OAUTHBEARER sasl mechanism, which is working fine in java/kotlin, also work fine in .Net. Next step is to implement same for <code>librdkafka</code> because we need it for Clickhouse DB's implementation of kafka connector which uses <code>librdkafka</code>.</p>
  <p>I tried with following impelenation for plugin:</p>
  <h3>custom-oauth.c</h3>
  <pre><code>#include &lt;librdkafka/rdkafka.h&gt;
  #include &lt;stdio.h&gt;
  
  void oauth_refresh (rd_kafka_t *rk, const char *oauthbearer_config, void *opaque) {
          char err_msg[2048];
          const char *token;
          long long expiry = 0;
          printf(&quot;============== Initializing oauthbearer config: %s\n&quot;, oauthbearer_config);
          token = &quot;&lt;custom logic implementation&gt;&quot;;
          rd_kafka_oauthbearer_set_token(rk, token, expiry, &quot;&quot;, NULL, 0, err_msg, sizeof(err_msg));
          printf(&quot;============== Oauthbearer token set: %s\n&quot;, token);
  }
  
  void conf_init(rd_kafka_conf_t *conf, const char *path, char *errstr, size_t errstr_size) {
          printf(&quot;============== registering callback\n&quot;);
          rd_kafka_conf_set_oauthbearer_token_refresh_cb(conf, oauth_refresh);
          if (rd_kafka_last_error()) {
                  printf(&quot;============== got some error: %d\n&quot;, rd_kafka_last_error());
          }
          printf(&quot;============== callback registered\n&quot;);
  }
  
  </code></pre>
  <p>Compile it with:</p>
  <pre><code>gcc -c -fPIC custom-oauth.c -o oauth-cb.o
  </code></pre>
  <p>Make shared library:</p>
  <pre><code>gcc -shared oauth-cb.o -lrdkafka -o lib-oauth-cb.so
  </code></pre>
  <p>Try it out with <code>kafkacat</code> by:</p>
  <pre><code>kafkacat -L \
      -X debug=&quot;all&quot; \
      -X log_level=0 \
      -b my-broker-1:9094 \
      -X security.protocol=SASL_SSL \
      -X ssl.ca.location=/path-to-my-cer.pem \
      -X sasl.mechanisms=OAUTHBEARER \
      -X plugin.library.paths=lib-oauth-cb.so \
      -X sasl.oauthbearer.config=&quot;my-custom-oauth-config&quot;
  </code></pre>
  <p>Results in:</p>
  <pre><code>============== registering callback
  ============== callback registered
  % ERROR: No such configuration property: &quot;plugin.library.paths&quot; (plugin lib-oauth-cb.so)
  </code></pre>
  <p>If I comment-out all printf-s happening after <code>rd_kafka_conf_set_oauthbearer_token_refresh_cb</code> then I don't get <code>% ERROR: No such configuration property: &quot;plugin.library.paths&quot; (plugin lib-oauth-cb.so)</code></p>
  <p>But then also doesn't work and my callback for setting oauth bearer token is never called.
  This is output:</p>
  <pre><code>============== registering callback
  %7|1605365045.759|SASL|rdkafka#producer-1| [thrd:app]: Selected provider OAUTHBEARER (builtin) for SASL mechanism OAUTHBEARER
  %7|1605365045.759|OPENSSL|rdkafka#producer-1| [thrd:app]: Using OpenSSL version OpenSSL 1.1.1g  21 Apr 2020 (0x1010107f, librdkafka built with 0x1010107f)
  %7|1605365045.762|SSL|rdkafka#producer-1| [thrd:app]: Loading CA certificate(s) from file /path-to-my-cer.pem
  %7|1605365045.763|WAKEUPFD|rdkafka#producer-1| [thrd:app]: sasl_ssl://my-broker-1:9094/bootstrap: Enabled low-latency ops queue wake-ups
  %7|1605365045.763|BRKMAIN|rdkafka#producer-1| [thrd::0/internal]: :0/internal: Enter main broker thread
  %7|1605365045.763|BROKER|rdkafka#producer-1| [thrd:app]: sasl_ssl://my-broker-1:9094/bootstrap: Added new broker with NodeId -1
  %7|1605365045.763|BRKMAIN|rdkafka#producer-1| [thrd:sasl_ssl://my-broker-1:9094/bootstrap]: sasl_ssl://my-broker-1:9094/bootstrap: Enter main broker thread
  %7|1605365045.763|CONNECT|rdkafka#producer-1| [thrd:app]: sasl_ssl://my-broker-1:9094/bootstrap: Selected for cluster connection: bootstrap servers added (broker has 0 connection attempt(s))
  %7|1605365045.763|CONNECT|rdkafka#producer-1| [thrd:sasl_ssl://my-broker-1:9094/bootstrap]: sasl_ssl://my-broker-1:9094/bootstrap: Received CONNECT op
  %7|1605365045.763|STATE|rdkafka#producer-1| [thrd:sasl_ssl://my-broker-1:9094/bootstrap]: sasl_ssl://my-broker-1:9094/bootstrap: Broker changed state INIT -&gt; TRY_CONNECT
  %7|1605365045.763|BROADCAST|rdkafka#producer-1| [thrd:sasl_ssl://my-broker-1:9094/bootstrap]: Broadcasting state change
  %7|1605365045.763|INIT|rdkafka#producer-1| [thrd:app]: librdkafka v1.5.0 (0x10500ff) rdkafka#producer-1 initialized (builtin.features gzip,snappy,ssl,sasl,regex,lz4,sasl_gssapi,sasl_plain,sasl_scram,plugins,zstd,sasl_oauthbearer, CC CXX PKGCONFIG OSXLD LIBDL PLUGINS ZLIB SSL SASL_CYRUS ZSTD HDRHISTOGRAM LZ4_EXT SYSLOG SNAPPY SOCKEM SASL_SCRAM SASL_OAUTHBEARER CRC32C_HW, debug 0xfffff)
  %7|1605365045.763|CONNECT|rdkafka#producer-1| [thrd:app]: Not selecting any broker for cluster connection: still suppressed for 49ms: application metadata request
  %7|1605365046.763|CONNECT|rdkafka#producer-1| [thrd:main]: Cluster connection already in progress: no cluster connection
  %7|1605365047.763|CONNECT|rdkafka#producer-1| [thrd:main]: Cluster connection already in progress: no cluster connection
  %7|1605365048.763|CONNECT|rdkafka#producer-1| [thrd:main]: Cluster connection already in progress: no cluster connection
  %7|1605365049.764|CONNECT|rdkafka#producer-1| [thrd:main]: Cluster connection already in progress: no cluster connection
  %7|1605365050.766|CONNECT|rdkafka#producer-1| [thrd:main]: Cluster connection already in progress: no cluster connection
  %7|1605365050.766|CONNECT|rdkafka#producer-1| [thrd:app]: Not selecting any broker for cluster connection: still suppressed for 49ms: application metadata request
  % ERROR: Failed to acquire metadata: Local: Broker transport failure
  </code></pre>
  <h2>Notes:</h2>
  <ul>
  <li>this kafka cluster has 3 different listeners
  <ul>
  <li>port 9092 PLAINTEXT protocol (tried kafkacat and it works)</li>
  <li>port 9093 SSL protocol (tried kafkacat and it works)</li>
  <li>port 9094 SASL_SSL protocol with OAUTHBEARER as sasl mechanism</li>
  </ul>
  </li>
  <li>I have installed <code>librdkafka</code> and <code>kafkacat</code> using <code>brew</code> on my Mac and as fas as I can see version of librdkafka is <code>1.5.0</code></li>
  </ul>
  <h1>Question</h1>
  <ol>
  <li><p>Why is my callback <code>oauth_refresh</code> never called (no <code>============== Initializing oauthbearer config</code>... printed on <code>stdout</code>)</p>
  </li>
  <li><p>Why do I get an error <code>No such configuration property: &quot;plugin.library.paths&quot; (plugin lib-oauth-cb.so)</code> if there is some <code>printf</code> after <code>rd_kafka_conf_set_oauthbearer_token_refresh_cb</code></p>
  <ul>
  <li>There won't be this error as long as <code>rd_kafka_conf_set_oauthbearer_token_refresh_cb</code> is called after <code>printf</code></li>
  <li>Even multiple times setting does not result in this error:
  <ul>
  <li><code>printf</code></li>
  <li><code>rd_kafka_conf_set_oauthbearer_token_refresh_cb</code></li>
  <li><code>printf</code></li>
  <li><code>rd_kafka_conf_set_oauthbearer_token_refresh_cb</code></li>
  </ul>
  </li>
  <li>What is the side effect of <code>printf</code> to cause this error?</li>
  <li>This part of question is less important, because I don't plan to keep those printf-s, part <code>1)</code> matters</li>
  </ul>
  </li>
  </ol>
  <h2>Goal</h2>
  <ul>
  <li>My goal is to have small pluggable lib of our OAUTHBEARER scheme which can be plugged in into anything which uses <code>librdkafka</code> (end goal is <a href="https://clickhouse.tech/docs/en/engines/table-engines/integrations/kafka/" rel="nofollow noreferrer">Clickhouse</a>'s kafka connector)</li>
  <li>I used <code>kafkacat</code> just because it's easy tool to experiment with</li>
  <li>Key point is plugability so that there is no need to recompile some app which uses librdkafka just to be able to use custom refresh oauth token logic.</li>
  </ul>
  <h1>Edit</h1>
  <p>I've had several issues:</p>
  <ul>
  <li>signature of conf_init:
  <ul>
  <li>needs to be: <code>rd_kafka_resp_err_t conf_init(rd_kafka_conf_t *conf, const char *path, char *errstr, size_t errstr_size)</code></li>
  <li>instead of: <code>void conf_init(rd_kafka_conf_t *conf, const char *path, char *errstr, size_t errstr_size)</code></li>
  </ul>
  </li>
  <li>when setting oauthbearer token, expiry needs to be <code>nowMs + someTTL</code> instead <code>0</code> in my example</li>
  <li>I ended up using <a href="https://github.com/edenhill/librdkafka/blob/master/examples/rdkafka_example.c" rel="nofollow noreferrer">rdkafka_example.c</a>  instead of installed <code>kafkacat</code> since latest version of kafka cat that i installed with <code>brew</code> was <code>1.5.0</code>
  <ul>
  <li>when running against this older <code>kafkacat</code> I was getting following error <code>kafkacat(76875,0x11b279dc0) malloc: *** error for object 0x10bad7bd0: pointer being freed was not allocated kafkacat(76875,0x11b279dc0) malloc: *** set a breakpoint in malloc_error_break to debug </code></li>
  </ul>
  </li>
  <li>I got it to work say in consumer mode <code>-C</code> but not in print metadata mode <code>-L</code> because with <code>-L</code> mode does not call <code>rd_kafka_poll</code> so my callback was never called.
  <ul>
  <li>I ended up adding <code>rd_kafka_poll(rk, 5000);</code> in <a href="https://github.com/edenhill/librdkafka/blob/master/examples/rdkafka_example.c" rel="nofollow noreferrer">rdkafka_example.c</a> when <code>-L</code> switch is enabled</li>
  </ul>
  </li>
  </ul>