PHP_METHOD(RdKafka__Kafka, __destruct)
{
    struct timeval tv;
    gettimeofday(&tv,NULL);
    fprintf(stdout, "%lu %lu Starting destructor.\n", (unsigned long)tv.tv_sec, (unsigned long)tv.tv\
_usec);
    kafka_object *intern;

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "") == FAILURE) {
        return;
    }

    intern = get_custom_object_zval(kafka_object, getThis());
    if (intern->rk) {
        int i = 0;
        while (rd_kafka_outq_len(intern->rk) > 0) {
            gettimeofday(&tv,NULL);
            fprintf(stderr, "%lu %lu waiting for queue to get empty.\n", (unsigned long) tv.tv_sec, \
(unsigned long)tv.tv_usec);
            i++;
            rd_kafka_poll(intern->rk, 50);
        }

        gettimeofday(&tv,NULL);
        fprintf(stderr, "%lu %lu destroying intern rk.\n", (unsigned long) tv.tv_sec, (unsigned long\
)tv.tv_usec);
        rd_kafka_destroy(intern->rk);
        intern->rk = NULL;
        gettimeofday(&tv,NULL);
        fprintf(stderr, "%lu %lu Leaving.\n", (unsigned long) tv.tv_sec, (unsigned long)tv.tv_usec);
    }
    kafka_conf_callbacks_dtor(&intern->cbs TSRMLS_CC);
}