def METHOD_NAME(self, mock_w_callback, mock_re):
    callback = _KafkaCallback(
        SinkReport(), record_envelope=mock_re, write_callback=mock_w_callback
    )
    mock_error = MagicMock()
    mock_message = MagicMock()
    callback.kafka_callback(mock_error, mock_message)
    mock_w_callback.on_failure.assert_called_once()
    assert mock_w_callback.on_failure.call_args[0][0] == mock_re
    assert mock_w_callback.on_failure.call_args[0][1] == mock_error
    callback.kafka_callback(None, mock_message)
    mock_w_callback.on_success.assert_called_once()
    assert mock_w_callback.on_success.call_args[0][0] == mock_re