def init_transactions(self):
        """
        Needs to be called before any other methods when the transactional.id is set in the configuration.

        This method does the following:
          1. Ensures any transactions initiated by previous instances of the producer with the same
             transactional_id are completed. If the previous instance had failed with a transaction in
             progress, it will be aborted. If the last transaction had begun completion,
             but not yet finished, this method awaits its completion.
          2. Gets the internal producer id and epoch, used in all future transactional
             messages issued by the producer.

        Note that this method will raise KafkaTimeoutError if the transactional state cannot
        be initialized before expiration of `max_block_ms`.

        Retrying after a KafkaTimeoutError will continue to wait for the prior request to succeed or fail.
        Retrying after any other exception will start a new initialization attempt.
        Retrying after a successful initialization will do nothing.

        Raises:
            IllegalStateError: if no transactional_id has been configured
            AuthorizationError: fatal error indicating that the configured
                transactional_id is not authorized.
            KafkaError: if the producer has encountered a previous fatal error or for any other unexpected error
            KafkaTimeoutError: if the time taken for initialize the transaction has surpassed `max.block.ms`.
        """
        if not self._transaction_manager:
            raise Errors.IllegalStateError("Cannot call init_transactions without setting a transactional_id.")
        if self._init_transactions_result is None:
            self._init_transactions_result = self._transaction_manager.initialize_transactions()
            self._sender.wakeup()

        try:
            if not self._init_transactions_result.wait(timeout_ms=self.config['max_block_ms']):
                raise Errors.KafkaTimeoutError("Timeout expired while initializing transactional state in %s ms." % (self.config['max_block_ms'],))
        finally:
            if self._init_transactions_result.failed:
                self._init_transactions_result = None

        Needs to be called before any other methods when the transactional.id is set in the configuration.

        This method does the following:
          1. Ensures any transactions initiated by previous instances of the producer with the same
             transactional_id are completed. If the previous instance had failed with a transaction in
             progress, it will be aborted. If the last transaction had begun completion,
             but not yet finished, this method awaits its completion.
          2. Gets the internal producer id and epoch, used in all future transactional
             messages issued by the producer.

        Note that this method will raise KafkaTimeoutError if the transactional state cannot
        be initialized before expiration of `max_block_ms`.

        Retrying after a KafkaTimeoutError will continue to wait for the prior request to succeed or fail.
        Retrying after any other exception will start a new initialization attempt.
        Retrying after a successful initialization will do nothing.

        Raises:
            IllegalStateError: if no transactional_id has been configured
            AuthorizationError: fatal error indicating that the configured
                transactional_id is not authorized.
            KafkaError: if the producer has encountered a previous fatal error or for any other unexpected error
            KafkaTimeoutError: if the time taken for initialize the transaction has surpassed `max.block.ms`.
        
kafka/producer/kafka.py