text:
authors:
- tryggvib
redirect_from: /2013/06/New-Features-in-May-2013/
title: New Features in May 2013
---

![Image by Mosman Library (cc-by 2.0)](https://farm9.staticflickr.com/8205/8206356927_3d5c720658_c.jpg "Badges!")

This is perhaps not a very attention grabbing title for this post, but some cool new features have landed in OpenSpending this month. We're really proud of them so we thought we'd share them with you.

### Badges

We've got badges! OpenSpending administrators can now award badges to outstanding datasets. To begin with there are only a few badges but we foresee collaboration with organisations that want to give their approval to datasets. These organisations can do quality assurance on datasets originating from them but have been modified to fit better into OpenSpending (the badge would then say: *"Yes this is still the data we published"*). These organisations could also give badges to datasets that help their cause (the badge would then say: *"This dataset helps us reach our goal"*).

If you're representing an organisation and want to be able to give out badges, please [get in touch](http://openspending.org/about/contact.html).

If you're managing a dataset, find out which [badges you can get](http://openspending.org/badges) and start collecting!

### Show Analysis Results

After loading a source users can't really see if the load was successful. It may appear to be successful but that might be just because OpenSpending was able to download something.

To provide some kind of feedback OpenSpending now shows the results of the analysis it does on the data. In particular it shows the columns that it found. If there aren't any columns or the column names are weird, users might now catch it before something goes wrong.

### EU Cookie Compliance

This may not be a big thing for you and it might even be slightly irritating but for us this is really important. We want to follow the law and OpenSpending now does so by implementing the EU Cookie Directive (if you want to know about cookies being placed on your computer this is important to you too).

Users are now presented with a small banner that tells them about the cookies and offers them link to a page where they can read more about [our cookie policy](http://okfn.org/cookie-policy/).

### OpenSpending Contacts Map

There are many projects out in the world that work in some way with spending data. We want to be able to connect those initiatives together. They can attract new contributors, learn from other projects and spark new interesting projects (even in other countries).

To help you establish these connections or to find projects you are interested in we've put up an OpenSpending app on [apps.openspending.org/oscontactsmap](http://apps.openspending.org/oscontactsmap/). There you can see the world and find all of the projects that relate to OpenSpending in some way.

If your project isn't there, don't panic! Thera are lot of projects so we might have missed some of you. Please let us know about your project and we'll add it to the list right away. When we're sure we have most of the projects on the list we'll make this map more prominent on OpenSpending's main site.

### Satellite Template

Before May users had to fork [Where Does My Money Go?](http://wheredoesmymoneygo.org/) or some other site in order to create their own satellite site (a site that provides context to the data and analysis in OpenSpending).

In May we created a satellite site template so you can easily recreate sites like *Where Does My Money Go?* with a simple config file without having to remove a lot of context specific information.

Just fork [our satellite template repository](https://github.com/openspending/satellite-template) and start configuring your own satellite site.

### Other Changes

There are loads of other smaller changes we did this month. We went through all of the issues in our issue tracker that were labelled as *bug* and fixed them. We tried to make some instructions clearer. We made small headway in getting better IE7 compatibility and we now show the type of a dimension that has been created with the model editor.

### Want to contribute? We need code reviewers!

We're really proud of what we've achieved this month. There are a lot of upcoming changes in the pipelines (in the form of pull requests). Quality assurance is really important to OpenSpending so we don't add anything new to the platform unless it's been looked at by at least one other developer (this is called code review).

If you're interested in contributing to OpenSpending, code reviews are a good place to start. You get familiar with the code base and you can (and should) raise all kinds of questions (so if in doubt about anything, just ask the developers).

There's even a law in open source software (Linus' Law): *"Given enough eyeballs, all bugs are shallow"*. That means the more people do code review, the better OpenSpending will get. That's why we'd appreciate your help in getting some more quality code into OpenSpending. Look at the pull requests in our repositories on GitHub and comment, both if you see something fishy and if you don't!

**Interested in being a code reviewer? [Get in touch!](http://openspending.org/about/contact.html)**

###Data Clinics and Community Hangouts - going forward
A lot of great spending projects in the making are in the need of community support for tips on how to clean and upload the data.

For this reason we'll therefore begin to run bi-weekly Data Clinics - a community drop in for all kinds of technical troubleshooting you need in order to get working on OpenSpending.
All you need to bring is your messy data, and you'll be able to ask the community on tips on how to tackle your spending data.

The first Data Clinic will take place **today Wednesday 5 June at 19:00 CET / 18:00 BST / 13:00 EDT** as chat via skype (ping: anpehej) or IRC #openspending. [Sign up here!](https://docs.google.com/a/okfn.org/document/d/1vx9oSJieuqfQ2xdBTyFSTmDRsPRenTrJLKJDo_ZP8z4/edit#)

With the introduction of the Data Clinic we'll now offering community activities every Wednesday - either as Data Clinic or Community GoogleHangout. We hope that you'll join some of the activities!

###Current community projects
Here are some of the projects currently in the works. Many of them could need a helping hand:
- Felix is working on Andalusian procurement data. Data quality seems good, and would be interesting to compare with the data available in OpenTED.
- Cecilia and the investigative journalism team at IRPI has claimed the Italian structural fund data and will be working to add it to OpenSpending. They need help cleaning a few of the data fields so please get in touch if you can help!
- Tarek would be interested looking at spending data from Egypt
- Pierre has helped scrape, clean and upload procurement data from Senegal to OpenSpending. He'll post on the OpenSpending blog soon.
- Benjamin is working on the EU procurement data opendted.pudo.org
- Transactional spending data from the city of [Washington DC](http://openspending.org/dc-vendors-contractors) was added during theweekend as part of #Hackforchange.

###Global Weekend on city spending data July 20-21
The past months has seen strong growth in local spending projects with [86 cities mapped](http://apps.openspending.org/maps/). As we want to support and grow these initiatives we're asking if you'll join us to organise a global spending weekend to open city spending on July 20-21. The spending weekend will offer online trainings and support to your local group to help you get your project off the ground.
Hiroichi (OKF Japan), Mor (Hasadna) and Adam (Open Budget Oakland) have already expressed interest in the idea, and we'd love to hear if you want to be involved.

Have a look and add your self to the [planning document](https://docs.google.com/a/okfn.org/document/d/1Zh-TPxgMiFDrzk-rNJqL9CmCbbtlZmp2xjWlZ6T20TA/edit#heading=h.6416s7qbnyrt).

### Thanks

Thanks to **Nigel Babu**, **Lucy Chambers**, **Martin Keegan**, **Anders Pedersen**, **Rufus Pollock**, **Stephen Russett**, and **Stefan Wehrmeyer** for their contributions this month (there are probably a lot more who've contributed somehow to this month's features so don't be sad if we forgot you, let us know and we'll add you).

<small>Image of badges used for this blog posts is by [Mosman Library](https://www.flickr.com/photos/mosmanlibrary/) on [Flickr](http://flickr.com), released under [Creative Commons Attribution, version 2.0](http://creativecommons.org/licenses/by/2.0/).</small>


text:
package org.apache.asterix.external.feed.watch;

import org.apache.asterix.external.feed.api.IExceptionHandler;
import org.apache.asterix.external.feed.api.IFeedMetricCollector;
import org.apache.asterix.external.feed.api.IFrameEventCallback;
import org.apache.asterix.external.feed.api.IFramePostProcessor;
import org.apache.asterix.external.feed.api.IFramePreprocessor;
import org.apache.asterix.external.feed.dataflow.FeedRuntimeInputHandler;
import org.apache.asterix.external.feed.management.FeedConnectionId;
import org.apache.asterix.external.feed.policy.FeedPolicyAccessor;
import org.apache.asterix.external.feed.runtime.FeedRuntimeId;
import org.apache.hyracks.api.comm.IFrameWriter;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;

public class ComputeSideMonitoredBuffer extends MonitoredBuffer {

    public ComputeSideMonitoredBuffer(IHyracksTaskContext ctx, FeedRuntimeInputHandler inputHandler, IFrameWriter frameWriter,
            FrameTupleAccessor fta, RecordDescriptor recordDesc, IFeedMetricCollector metricCollector,
            FeedConnectionId connectionId, FeedRuntimeId runtimeId, IExceptionHandler exceptionHandler,
            IFrameEventCallback callback, int nPartitions, FeedPolicyAccessor policyAccessor) {
        super(ctx, inputHandler, frameWriter, fta, recordDesc, metricCollector, connectionId, runtimeId,
                exceptionHandler, callback, nPartitions, policyAccessor);
    }

    @Override
    protected boolean monitorProcessingRate() {
        return true;
    }

    protected boolean logInflowOutflowRate() {
        return true;
    }

    @Override
    protected boolean monitorInputQueueLength() {
        return true;
    }

    @Override
    protected IFramePreprocessor getFramePreProcessor() {
        return null;
    }

    @Override
    protected IFramePostProcessor getFramePostProcessor() {
        return null;
    }

    @Override
    protected boolean reportOutflowRate() {
        return false;
    }

    @Override
    protected boolean reportInflowRate() {
        return false;
    }

}

redpajama_set_name: RedPajamaGithub


text:
import $ from "jquery";

class Requester {

    get(url, headers) {
        var promise = new Promise(function(resolve, reject) {
            $.ajax({
                url,
                method: "GET",
                headers,
                contentType: "application/json",
                success(response) {
                    resolve(response);
                },
                error(err) {
                    reject(err);
                }
            });
        });

        return promise;
    }

    post(url, headers, data) {
        var promise = new Promise(function(resolve, reject) {
            $.ajax({
                url,
                method: "POST",
                headers,
                data: JSON.stringify(data),
                contentType: "application/json",
                success: function(data) {
                    resolve(data);
                },
                error: function(err) {
                    reject(err);
                },
            });
        });

        return promise;
    }

    put(url, headers, data) {
        var promise = new Promise(function(resolve, reject) {
            $.ajax({
                url,
                method: "PUT",
                headers,
                data: JSON.stringify(data),
                contentType: "application/json",
                success: function(data) {
                    resolve(data);
                },
                error: function(err) {
                    reject(err);
                },
            });
        });

        return promise;
    }
}

let requester = new Requester();
export { requester };


text:
<?php
namespace tests\app\decibel\utility;

use app\decibel\regional\DLabel;
use app\decibel\test\DTestCase;
use app\decibel\utility\DBasicDefinable;

class TestBasicDefinable
{
    use DBasicDefinable;
    protected $property1 = 'default1';
    protected $property2 = 'default2';

    public function setProperty2($value)
    {
        $this->property2 = $value;

        return $this;
    }

    public function getProperty2()
    {
        return $this->property2;
    }

    public function testSetArray($name, $value, $allowNull = false, $expected = null)
    {
        return $this->setArray($name, $value, $allowNull, $expected);
    }

    public function testSetBoolean($name, $value, $allowNull = false, $expected = null)
    {
        return $this->setBoolean($name, $value, $allowNull, $expected);
    }

    public function testSetEnum($name, $value, array $values, $allowNull = false, $expected = null)
    {
        return $this->setEnum($name, $value, $values, $allowNull, $expected);
    }

    public function testSetInteger($name, $value, $allowNull = false, $expected = null)
    {
        return $this->setInteger($name, $value, $allowNull, $expected);
    }

    public function testSetLabel($name, $value, $allowString = true)
    {
        return $this->setLabel($name, $value, $allowString);
    }
}

/**
 * Test class for DStringField.
 * Generated by Decibel on 2012-04-12 at 09:08:41.
 */
class DStringFieldTest extends DTestCase
{
    /**
     * @covers app\decibel\utility\DBasicDefinable::__get
     */
    public function test__get()
    {
        $definable = new TestBasicDefinable();
        $this->assertSame('default1', $definable->property1);
        $this->assertSame('default2', $definable->property2);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::__get
     * @expectedException app\decibel\debug\DInvalidPropertyException
     */
    public function test__get_invalid()
    {
        $definable = new TestBasicDefinable();
        $definable->property3;
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::__set
     */
    public function test__set()
    {
        $definable = new TestBasicDefinable();
        $definable->property1 = 'test1';
        $definable->property2 = 'test2';
        $this->assertSame('test1', $definable->property1);
        $this->assertSame('test2', $definable->property2);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::__set
     * @expectedException app\decibel\debug\DInvalidPropertyException
     */
    public function test__set_invalid()
    {
        $definable = new TestBasicDefinable();
        $definable->property3 = 'test';
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setArray
     */
    public function testsetArray()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetArray('property1', array(1, 2, 3));
        $this->assertSame(array(1, 2, 3), $definable->property1);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setArray
     * @expectedException app\decibel\debug\DInvalidParameterValueException
     */
    public function testsetArray_invalid()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetArray('property1', 1);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setArray
     * @expectedException app\decibel\debug\DInvalidParameterValueException
     */
    public function testsetArray_invalid_allowNull()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetArray('property1', 1, true);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setArray
     * @expectedException app\decibel\debug\DInvalidParameterValueException
     */
    public function testsetArray_invalidNull()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetArray('property1', null);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setArray
     */
    public function testsetArray_null()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetArray('property1', null, true);
        $this->assertNull($definable->property1);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setBoolean
     */
    public function testsetBoolean()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetBoolean('property1', true);
        $this->assertTrue($definable->property1);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setBoolean
     */
    public function testsetBoolean_null()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetBoolean('property1', null, true);
        $this->assertNull($definable->property1);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setBoolean
     * @expectedException app\decibel\debug\DInvalidParameterValueException
     */
    public function testsetBoolean_invalid()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetBoolean('property1', 100);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setBoolean
     * @expectedException app\decibel\debug\DInvalidParameterValueException
     */
    public function testsetBoolean_invalidNull()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetBoolean('property1', null);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setEnum
     */
    public function testsetEnum()
    {
        $values = array(
            'option1' => 'value1',
            'option2' => 'value2',
        );
        $definable = new TestBasicDefinable();
        $definable->testSetEnum('property1', 'option1', $values);
        $this->assertSame('option1', $definable->property1);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setEnum
     */
    public function testsetEnum_null()
    {
        $values = array(
            'option1' => 'value1',
            'option2' => 'value2',
        );
        $definable = new TestBasicDefinable();
        $definable->testSetEnum('property1', null, $values, true);
        $this->assertNull($definable->property1);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setEnum
     * @expectedException app\decibel\debug\DInvalidParameterValueException
     */
    public function testsetEnum_invalid()
    {
        $values = array(
            'option1' => 'value1',
            'option2' => 'value2',
        );
        $definable = new TestBasicDefinable();
        $definable->testSetEnum('property1', 100, $values);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setEnum
     * @expectedException app\decibel\debug\DInvalidParameterValueException
     */
    public function testsetEnum_invalidNull()
    {
        $values = array(
            'option1' => 'value1',
            'option2' => 'value2',
        );
        $definable = new TestBasicDefinable();
        $definable->testSetEnum('property1', null, $values);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setInteger
     */
    public function testsetInteger()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetInteger('property1', 10);
        $this->assertSame(10, $definable->property1);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setInteger
     */
    public function testsetInteger_null()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetInteger('property1', null, true);
        $this->assertNull($definable->property1);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setInteger
     * @expectedException app\decibel\debug\DInvalidParameterValueException
     */
    public function testsetInteger_invalid()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetInteger('property1', 10.1);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setInteger
     * @expectedException app\decibel\debug\DInvalidParameterValueException
     */
    public function testsetInteger_invalidNull()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetInteger('property1', null);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setLabel
     */
    public function testsetLabel()
    {
        $label = new DLabel('app\decibel', 'yes');
        $definable = new TestBasicDefinable();
        $definable->testSetLabel('property1', $label);
        $this->assertSame($label, $definable->property1);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setLabel
     */
    public function testsetLabel_string()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetLabel('property1', 'label');
        $this->assertSame('label', $definable->property1);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setLabel
     * @expectedException app\decibel\debug\DInvalidParameterValueException
     */
    public function testsetLabel_invalid()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetLabel('property1', 10);
    }

    /**
     * @covers app\decibel\utility\DBasicDefinable::setLabel
     * @expectedException app\decibel\debug\DInvalidParameterValueException
     */
    public function testsetLabel_invalidString()
    {
        $definable = new TestBasicDefinable();
        $definable->testSetLabel('property1', 'label', false);
    }
}


text:
package org.apache.flex.compiler.internal.css;

import static org.hamcrest.core.Is.is;
import static org.hamcrest.core.IsNot.not;
import static org.junit.Assert.assertThat;

import java.util.List;

import org.apache.flex.compiler.css.CombinatorType;
import org.apache.flex.compiler.css.ICSSCombinator;
import org.apache.flex.compiler.css.ICSSDocument;
import org.apache.flex.compiler.css.ICSSSelector;
import org.junit.Test;

/**
 * JUnit tests for {@link CSSCombinator}.
 *
 * @author Gordon Smith
 */
public class CSSCombinatorTests extends CSSBaseTests {

    private String getPostfix()
    {
    	return " { } ";
    }

	@Override
	public ICSSDocument getCSSNodeBase(String code) {
		return super.getCSSNodeBase(code + getPostfix());
	}

	protected CSSCombinator getCSSCombinator(String code) {
		List<ICSSSelector> selectors =  getCSSSelectors( code );
		assertThat("selectors", selectors,  not( (List<ICSSSelector>) null) );
		assertThat("selectors.size()", selectors.size(),  is( 1 ) );

		ICSSSelector selector = selectors.get(0);
		assertThat("selector", selector,  not( (ICSSSelector) null) );

		return (CSSCombinator) selector.getCombinator();
	}

	@Test
	public void CSSSelectorConditionTests_descendant_combinator1()
	{
		String code = " s|VBox s|Label ";

		CSSCombinator combinator = getCSSCombinator(code);
		//TODO why CSSCombinator doesn't extend CSSNodeBase??
		assertThat("combinator.getOperator()" , combinator.getOperator(), is( CSSModelTreeType.COMBINATOR ) );
		assertThat("combinator.getCombinatorType()" , combinator.getCombinatorType(), is(CombinatorType.DESCENDANT) );
		assertThat("combinator.getSelector()" , combinator.getSelector(), not( (ICSSSelector) null) );

		ICSSSelector selector = combinator.getSelector();
		assertThat("selector.getElementName()" , selector.getElementName(), is( "VBox" ) );
		assertThat("selector.getNamespacePrefix()" , selector.getNamespacePrefix(), is( "s" ) );
		assertThat("selector.getCombinator()" , selector.getCombinator(), is( (ICSSCombinator) null) );
	}

	@Test
	public void CSSSelectorConditionTests_descendant_combinator2()
	{
		String code = " s|VBox s|HBox s|Label";

		CSSCombinator combinator = getCSSCombinator(code);
		assertThat("combinator.getOperator()" , combinator.getOperator(), is( CSSModelTreeType.COMBINATOR ) );
		assertThat("combinator.getCombinatorType()" , combinator.getCombinatorType(), is(CombinatorType.DESCENDANT) );
		assertThat("combinator.getSelector()" , combinator.getSelector(), not( (ICSSSelector) null) );

		ICSSSelector selector1 = combinator.getSelector();
		assertThat("selector1.getElementName()" , selector1.getElementName(), is( "HBox" ) );
		assertThat("selector1.getNamespacePrefix()" , selector1.getNamespacePrefix(), is( "s" ) );
		assertThat("selector1.getCombinator()" , selector1.getCombinator(), not( (ICSSCombinator) null) );
		assertThat("selector1.getCombinator().getSelector()" , selector1.getCombinator().getSelector(), not( (ICSSSelector) null) );

		ICSSSelector selector2 = selector1.getCombinator().getSelector();
		assertThat("selector2.getElementName()" , selector2.getElementName(), is( "VBox" ) );
		assertThat("selector2.getNamespacePrefix()" , selector2.getNamespacePrefix(), is( "s" ) );
		assertThat("selector2.getCombinator()" , selector2.getCombinator(), is( (ICSSCombinator) null) );

	}

}


text:
package auth

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"strconv"
	"strings"

	"firebase.google.com/go/v4/internal"
	"google.golang.org/api/iterator"
)

// Tenant represents a tenant in a multi-tenant application.
//
// Multi-tenancy support requires Google Cloud's Identity Platform (GCIP). To learn more about GCIP,
// including pricing and features, see https://cloud.google.com/identity-platform.
//
// Before multi-tenancy can be used in a Google Cloud Identity Platform project, tenants must be
// enabled in that project via the Cloud Console UI.
//
// A tenant configuration provides information such as the display name, tenant identifier and email
// authentication configuration. For OIDC/SAML provider configuration management, TenantClient
// instances should be used instead of a Tenant to retrieve the list of configured IdPs on a tenant.
// When configuring these providers, note that tenants will inherit whitelisted domains and
// authenticated redirect URIs of their parent project.
//
// All other settings of a tenant will also be inherited. These will need to be managed from the
// Cloud Console UI.
type Tenant struct {
	ID                    string `json:"name"`
	DisplayName           string `json:"displayName"`
	AllowPasswordSignUp   bool   `json:"allowPasswordSignup"`
	EnableEmailLinkSignIn bool   `json:"enableEmailLinkSignin"`
	EnableAnonymousUsers  bool   `json:"enableAnonymousUser"`
}

// TenantClient is used for managing users, configuring SAML/OIDC providers, and generating email
// links for specific tenants.
//
// Before multi-tenancy can be used in a Google Cloud Identity Platform project, tenants must be
// enabled in that project via the Cloud Console UI.
//
// Each tenant contains its own identity providers, settings and users. TenantClient enables
// managing users and SAML/OIDC configurations of specific tenants. It also supports verifying ID
// tokens issued to users who are signed into specific tenants.
//
// TenantClient instances for a specific tenantID can be instantiated by calling
// [TenantManager.AuthForTenant(tenantID)].
type TenantClient struct {
	*baseClient
}

// TenantID returns the ID of the tenant to which this TenantClient instance belongs.
func (tc *TenantClient) TenantID() string {
	return tc.tenantID
}

// TenantManager is the interface used to manage tenants in a multi-tenant application.
//
// This supports creating, updating, listing, deleting the tenants of a Firebase project. It also
// supports creating new TenantClient instances scoped to specific tenant IDs.
type TenantManager struct {
	base       *baseClient
	endpoint   string
	projectID  string
	httpClient *internal.HTTPClient
}

func newTenantManager(client *internal.HTTPClient, conf *internal.AuthConfig, base *baseClient) *TenantManager {
	return &TenantManager{
		base:       base,
		endpoint:   base.tenantMgtEndpoint,
		projectID:  conf.ProjectID,
		httpClient: client,
	}
}

// AuthForTenant creates a new TenantClient scoped to a given tenantID.
func (tm *TenantManager) AuthForTenant(tenantID string) (*TenantClient, error) {
	if tenantID == "" {
		return nil, errors.New("tenantID must not be empty")
	}

	return &TenantClient{
		baseClient: tm.base.withTenantID(tenantID),
	}, nil
}

// Tenant returns the tenant with the given ID.
func (tm *TenantManager) Tenant(ctx context.Context, tenantID string) (*Tenant, error) {
	if tenantID == "" {
		return nil, errors.New("tenantID must not be empty")
	}

	req := &internal.Request{
		Method: http.MethodGet,
		URL:    fmt.Sprintf("/tenants/%s", tenantID),
	}
	var tenant Tenant
	if _, err := tm.makeRequest(ctx, req, &tenant); err != nil {
		return nil, err
	}

	tenant.ID = extractResourceID(tenant.ID)
	return &tenant, nil
}

// CreateTenant creates a new tenant with the given options.
func (tm *TenantManager) CreateTenant(ctx context.Context, tenant *TenantToCreate) (*Tenant, error) {
	if tenant == nil {
		return nil, errors.New("tenant must not be nil")
	}

	req := &internal.Request{
		Method: http.MethodPost,
		URL:    "/tenants",
		Body:   internal.NewJSONEntity(tenant.ensureParams()),
	}
	var result Tenant
	if _, err := tm.makeRequest(ctx, req, &result); err != nil {
		return nil, err
	}

	result.ID = extractResourceID(result.ID)
	return &result, nil
}

// UpdateTenant updates an existing tenant with the given options.
func (tm *TenantManager) UpdateTenant(ctx context.Context, tenantID string, tenant *TenantToUpdate) (*Tenant, error) {
	if tenantID == "" {
		return nil, errors.New("tenantID must not be empty")
	}
	if tenant == nil {
		return nil, errors.New("tenant must not be nil")
	}

	mask := tenant.params.UpdateMask()
	if len(mask) == 0 {
		return nil, errors.New("no parameters specified in the update request")
	}

	req := &internal.Request{
		Method: http.MethodPatch,
		URL:    fmt.Sprintf("/tenants/%s", tenantID),
		Body:   internal.NewJSONEntity(tenant.params),
		Opts: []internal.HTTPOption{
			internal.WithQueryParam("updateMask", strings.Join(mask, ",")),
		},
	}
	var result Tenant
	if _, err := tm.makeRequest(ctx, req, &result); err != nil {
		return nil, err
	}

	result.ID = extractResourceID(result.ID)
	return &result, nil
}

// DeleteTenant deletes the tenant with the given ID.
func (tm *TenantManager) DeleteTenant(ctx context.Context, tenantID string) error {
	if tenantID == "" {
		return errors.New("tenantID must not be empty")
	}

	req := &internal.Request{
		Method: http.MethodDelete,
		URL:    fmt.Sprintf("/tenants/%s", tenantID),
	}
	_, err := tm.makeRequest(ctx, req, nil)
	return err
}

// Tenants returns an iterator over tenants in the project.
//
// If nextPageToken is empty, the iterator will start at the beginning. Otherwise,
// iterator starts after the token.
func (tm *TenantManager) Tenants(ctx context.Context, nextPageToken string) *TenantIterator {
	it := &TenantIterator{
		ctx: ctx,
		tm:  tm,
	}
	it.pageInfo, it.nextFunc = iterator.NewPageInfo(
		it.fetch,
		func() int { return len(it.tenants) },
		func() interface{} { b := it.tenants; it.tenants = nil; return b })
	it.pageInfo.MaxSize = maxConfigs
	it.pageInfo.Token = nextPageToken
	return it
}

func (tm *TenantManager) makeRequest(ctx context.Context, req *internal.Request, v interface{}) (*internal.Response, error) {
	if tm.projectID == "" {
		return nil, errors.New("project id not available")
	}

	req.URL = fmt.Sprintf("%s/projects/%s%s", tm.endpoint, tm.projectID, req.URL)
	return tm.httpClient.DoAndUnmarshal(ctx, req, v)
}

const (
	tenantDisplayNameKey     = "displayName"
	allowPasswordSignUpKey   = "allowPasswordSignup"
	enableEmailLinkSignInKey = "enableEmailLinkSignin"
	enableAnonymousUser      = "enableAnonymousUser"
)

// TenantToCreate represents the options used to create a new tenant.
type TenantToCreate struct {
	params nestedMap
}

// DisplayName sets the display name of the new tenant.
func (t *TenantToCreate) DisplayName(name string) *TenantToCreate {
	return t.set(tenantDisplayNameKey, name)
}

// AllowPasswordSignUp enables or disables email sign-in provider.
func (t *TenantToCreate) AllowPasswordSignUp(allow bool) *TenantToCreate {
	return t.set(allowPasswordSignUpKey, allow)
}

// EnableEmailLinkSignIn enables or disables email link sign-in.
//
// Disabling this makes the password required for email sign-in.
func (t *TenantToCreate) EnableEmailLinkSignIn(enable bool) *TenantToCreate {
	return t.set(enableEmailLinkSignInKey, enable)
}

// EnableAnonymousUsers enables or disables anonymous authentication.
func (t *TenantToCreate) EnableAnonymousUsers(enable bool) *TenantToCreate {
	return t.set(enableAnonymousUser, enable)
}

func (t *TenantToCreate) set(key string, value interface{}) *TenantToCreate {
	t.ensureParams().Set(key, value)
	return t
}

func (t *TenantToCreate) ensureParams() nestedMap {
	if t.params == nil {
		t.params = make(nestedMap)
	}

	return t.params
}

// TenantToUpdate represents the options used to update an existing tenant.
type TenantToUpdate struct {
	params nestedMap
}

// DisplayName sets the display name of the new tenant.
func (t *TenantToUpdate) DisplayName(name string) *TenantToUpdate {
	return t.set(tenantDisplayNameKey, name)
}

// AllowPasswordSignUp enables or disables email sign-in provider.
func (t *TenantToUpdate) AllowPasswordSignUp(allow bool) *TenantToUpdate {
	return t.set(allowPasswordSignUpKey, allow)
}

// EnableEmailLinkSignIn enables or disables email link sign-in.
//
// Disabling this makes the password required for email sign-in.
func (t *TenantToUpdate) EnableEmailLinkSignIn(enable bool) *TenantToUpdate {
	return t.set(enableEmailLinkSignInKey, enable)
}

// EnableAnonymousUsers enables or disables anonymous authentication.
func (t *TenantToUpdate) EnableAnonymousUsers(enable bool) *TenantToUpdate {
	return t.set(enableAnonymousUser, enable)
}

func (t *TenantToUpdate) set(key string, value interface{}) *TenantToUpdate {
	if t.params == nil {
		t.params = make(nestedMap)
	}

	t.params.Set(key, value)
	return t
}

// TenantIterator is an iterator over tenants.
type TenantIterator struct {
	tm       *TenantManager
	ctx      context.Context
	nextFunc func() error
	pageInfo *iterator.PageInfo
	tenants  []*Tenant
}

// PageInfo supports pagination.
func (it *TenantIterator) PageInfo() *iterator.PageInfo {
	return it.pageInfo
}

// Next returns the next Tenant. The error value of [iterator.Done] is
// returned if there are no more results. Once Next returns [iterator.Done], all
// subsequent calls will return [iterator.Done].
func (it *TenantIterator) Next() (*Tenant, error) {
	if err := it.nextFunc(); err != nil {
		return nil, err
	}

	tenant := it.tenants[0]
	it.tenants = it.tenants[1:]
	return tenant, nil
}

func (it *TenantIterator) fetch(pageSize int, pageToken string) (string, error) {
	params := map[string]string{
		"pageSize": strconv.Itoa(pageSize),
	}
	if pageToken != "" {
		params["pageToken"] = pageToken
	}

	req := &internal.Request{
		Method: http.MethodGet,
		URL:    "/tenants",
		Opts: []internal.HTTPOption{
			internal.WithQueryParams(params),
		},
	}

	var result struct {
		Tenants       []Tenant `json:"tenants"`
		NextPageToken string   `json:"nextPageToken"`
	}
	if _, err := it.tm.makeRequest(it.ctx, req, &result); err != nil {
		return "", err
	}

	for i := range result.Tenants {
		result.Tenants[i].ID = extractResourceID(result.Tenants[i].ID)
		it.tenants = append(it.tenants, &result.Tenants[i])
	}

	it.pageInfo.Token = result.NextPageToken
	return result.NextPageToken, nil
}


text:
@class BSBserValues;
@class BSBserWriter;
@class IOSByteArray;

#define APRequestLoadDialogs_HEADER 104

@interface APRequestLoadDialogs : APRequest

#pragma mark Public

- (instancetype)init;

- (instancetype)initWithLong:(jlong)minDate
                     withInt:(jint)limit;

+ (APRequestLoadDialogs *)fromBytesWithByteArray:(IOSByteArray *)data;

- (jint)getHeaderKey;

- (jint)getLimit;

- (jlong)getMinDate;

- (void)parseWithBSBserValues:(BSBserValues *)values;

- (void)serializeWithBSBserWriter:(BSBserWriter *)writer;

- (NSString *)description;

@end

J2OBJC_EMPTY_STATIC_INIT(APRequestLoadDialogs)

J2OBJC_STATIC_FIELD_GETTER(APRequestLoadDialogs, HEADER, jint)

FOUNDATION_EXPORT APRequestLoadDialogs *APRequestLoadDialogs_fromBytesWithByteArray_(IOSByteArray *data);

FOUNDATION_EXPORT void APRequestLoadDialogs_initWithLong_withInt_(APRequestLoadDialogs *self, jlong minDate, jint limit);

FOUNDATION_EXPORT APRequestLoadDialogs *new_APRequestLoadDialogs_initWithLong_withInt_(jlong minDate, jint limit) NS_RETURNS_RETAINED;

FOUNDATION_EXPORT void APRequestLoadDialogs_init(APRequestLoadDialogs *self);

FOUNDATION_EXPORT APRequestLoadDialogs *new_APRequestLoadDialogs_init() NS_RETURNS_RETAINED;

J2OBJC_TYPE_LITERAL_HEADER(APRequestLoadDialogs)

typedef APRequestLoadDialogs ImActorModelApiRpcRequestLoadDialogs;

#endif // _APRequestLoadDialogs_H_


text:
from __future__ import unicode_literals

from django.db import migrations

from kolibri.core.content.utils.annotation import calculate_next_order

def calculate_channel_order(apps, schema_editor):
    ChannelMetadata = apps.get_model("content", "ChannelMetadata")
    for channel in ChannelMetadata.objects.all():
        calculate_next_order(channel)

class Migration(migrations.Migration):

    dependencies = [("content", "0015_auto_20190125_1715")]

    operations = [
        migrations.RunPython(calculate_channel_order, migrations.RunPython.noop)
    ]


text:
package kafka.consumer

import java.net.InetAddress
import java.util.UUID
import java.util.concurrent._
import java.util.concurrent.atomic._
import java.util.concurrent.locks.ReentrantLock

import com.yammer.metrics.core.Gauge
import kafka.api._
import kafka.client.ClientUtils
import kafka.cluster._
import kafka.common._
import kafka.javaapi.consumer.ConsumerRebalanceListener
import kafka.metrics._
import kafka.network.BlockingChannel
import kafka.serializer._
import kafka.utils.CoreUtils.inLock
import kafka.utils.ZkUtils._
import kafka.utils._
import org.I0Itec.zkclient.exception.ZkNodeExistsException
import org.I0Itec.zkclient.{IZkChildListener, IZkDataListener, IZkStateListener}
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.security.JaasUtils
import org.apache.zookeeper.Watcher.Event.KeeperState

import scala.collection._
import scala.collection.JavaConverters._

/**
 * This class handles the consumers interaction with zookeeper
 *
 * Directories:
 * 1. Consumer id registry:
 * /consumers/[group_id]/ids/[consumer_id] -> topic1,...topicN
 * A consumer has a unique consumer id within a consumer group. A consumer registers its id as an ephemeral znode
 * and puts all topics that it subscribes to as the value of the znode. The znode is deleted when the client is gone.
 * A consumer subscribes to event changes of the consumer id registry within its group.
 *
 * The consumer id is picked up from configuration, instead of the sequential id assigned by ZK. Generated sequential
 * ids are hard to recover during temporary connection loss to ZK, since it's difficult for the client to figure out
 * whether the creation of a sequential znode has succeeded or not. More details can be found at
 * (http://wiki.apache.org/hadoop/ZooKeeper/ErrorHandling)
 *
 * 2. Broker node registry:
 * /brokers/[0...N] --> { "host" : "host:port",
 *                        "topics" : {"topic1": ["partition1" ... "partitionN"], ...,
 *                                    "topicN": ["partition1" ... "partitionN"] } }
 * This is a list of all present broker brokers. A unique logical node id is configured on each broker node. A broker
 * node registers itself on start-up and creates a znode with the logical node id under /brokers. The value of the znode
 * is a JSON String that contains (1) the host name and the port the broker is listening to, (2) a list of topics that
 * the broker serves, (3) a list of logical partitions assigned to each topic on the broker.
 * A consumer subscribes to event changes of the broker node registry.
 *
 * 3. Partition owner registry:
 * /consumers/[group_id]/owner/[topic]/[broker_id-partition_id] --> consumer_node_id
 * This stores the mapping before broker partitions and consumers. Each partition is owned by a unique consumer
 * within a consumer group. The mapping is reestablished after each rebalancing.
 *
 * 4. Consumer offset tracking:
 * /consumers/[group_id]/offsets/[topic]/[broker_id-partition_id] --> offset_counter_value
 * Each consumer tracks the offset of the latest message consumed for each partition.
 *
 */
private[kafka] object ZookeeperConsumerConnector {
  val shutdownCommand: FetchedDataChunk = new FetchedDataChunk(null, null, -1L)
}

private[kafka] class ZookeeperConsumerConnector(val config: ConsumerConfig,
                                                val enableFetcher: Boolean) // for testing only
        extends ConsumerConnector with Logging with KafkaMetricsGroup {

  private val isShuttingDown = new AtomicBoolean(false)
  private val rebalanceLock = new Object
  private var fetcher: Option[ConsumerFetcherManager] = None
  private var zkUtils: ZkUtils = null
  private var topicRegistry = new Pool[String, Pool[Int, PartitionTopicInfo]]
  private val checkpointedZkOffsets = new Pool[TopicAndPartition, Long]
  private val topicThreadIdAndQueues = new Pool[(String, ConsumerThreadId), BlockingQueue[FetchedDataChunk]]
  private val scheduler = new KafkaScheduler(threads = 1, threadNamePrefix = "kafka-consumer-scheduler-")
  private val messageStreamCreated = new AtomicBoolean(false)

  private var sessionExpirationListener: ZKSessionExpireListener = null
  private var topicPartitionChangeListener: ZKTopicPartitionChangeListener = null
  private var loadBalancerListener: ZKRebalancerListener = null

  private var offsetsChannel: BlockingChannel = null
  private val offsetsChannelLock = new Object

  private var wildcardTopicWatcher: ZookeeperTopicEventWatcher = null
  private var consumerRebalanceListener: ConsumerRebalanceListener = null

  // useful for tracking migration of consumers to store offsets in kafka
  private val kafkaCommitMeter = newMeter("KafkaCommitsPerSec", "commits", TimeUnit.SECONDS, Map("clientId" -> config.clientId))
  private val zkCommitMeter = newMeter("ZooKeeperCommitsPerSec", "commits", TimeUnit.SECONDS, Map("clientId" -> config.clientId))
  private val rebalanceTimer = new KafkaTimer(newTimer("RebalanceRateAndTime", TimeUnit.MILLISECONDS, TimeUnit.SECONDS, Map("clientId" -> config.clientId)))

  newGauge(
    "yammer-metrics-count",
    new Gauge[Int] {
      def value = {
        com.yammer.metrics.Metrics.defaultRegistry().allMetrics().size()
      }
    }
  )

  val consumerIdString = {
    var consumerUuid : String = null
    config.consumerId match {
      case Some(consumerId) // for testing only
      => consumerUuid = consumerId
      case None // generate unique consumerId automatically
      => val uuid = UUID.randomUUID()
      consumerUuid = "%s-%d-%s".format(
        InetAddress.getLocalHost.getHostName, System.currentTimeMillis,
        uuid.getMostSignificantBits().toHexString.substring(0,8))
    }
    config.groupId + "_" + consumerUuid
  }
  this.logIdent = "[" + consumerIdString + "], "

  connectZk()
  createFetcher()
  ensureOffsetManagerConnected()

  if (config.autoCommitEnable) {
    scheduler.startup
    info("starting auto committer every " + config.autoCommitIntervalMs + " ms")
    scheduler.schedule("kafka-consumer-autocommit",
                       autoCommit,
                       delay = config.autoCommitIntervalMs,
                       period = config.autoCommitIntervalMs,
                       unit = TimeUnit.MILLISECONDS)
  }

  KafkaMetricsReporter.startReporters(config.props)
  AppInfo.registerInfo()

  def this(config: ConsumerConfig) = this(config, true)

  def createMessageStreams(topicCountMap: Map[String,Int]): Map[String, List[KafkaStream[Array[Byte],Array[Byte]]]] =
    createMessageStreams(topicCountMap, new DefaultDecoder(), new DefaultDecoder())

  def createMessageStreams[K,V](topicCountMap: Map[String,Int], keyDecoder: Decoder[K], valueDecoder: Decoder[V])
      : Map[String, List[KafkaStream[K,V]]] = {
    if (messageStreamCreated.getAndSet(true))
      throw new MessageStreamsExistException(this.getClass.getSimpleName +
                                   " can create message streams at most once",null)
    consume(topicCountMap, keyDecoder, valueDecoder)
  }

  def createMessageStreamsByFilter[K,V](topicFilter: TopicFilter,
                                        numStreams: Int,
                                        keyDecoder: Decoder[K] = new DefaultDecoder(),
                                        valueDecoder: Decoder[V] = new DefaultDecoder()) = {
    val wildcardStreamsHandler = new WildcardStreamsHandler[K,V](topicFilter, numStreams, keyDecoder, valueDecoder)
    wildcardStreamsHandler.streams
  }

  def setConsumerRebalanceListener(listener: ConsumerRebalanceListener) {
    if (messageStreamCreated.get())
      throw new MessageStreamsExistException(this.getClass.getSimpleName +
        " can only set consumer rebalance listener before creating streams",null)
    consumerRebalanceListener = listener
  }

  private def createFetcher() {
    if (enableFetcher)
      fetcher = Some(new ConsumerFetcherManager(consumerIdString, config, zkUtils))
  }

  private def connectZk() {
    info("Connecting to zookeeper instance at " + config.zkConnect)
    zkUtils = ZkUtils(config.zkConnect,
                      config.zkSessionTimeoutMs,
                      config.zkConnectionTimeoutMs,
                      JaasUtils.isZkSecurityEnabled())
  }

  // Blocks until the offset manager is located and a channel is established to it.
  private def ensureOffsetManagerConnected() {
    if (config.offsetsStorage == "kafka") {
      if (offsetsChannel == null || !offsetsChannel.isConnected)
        offsetsChannel = ClientUtils.channelToOffsetManager(config.groupId, zkUtils,
          config.offsetsChannelSocketTimeoutMs, config.offsetsChannelBackoffMs)

      debug("Connected to offset manager %s:%d.".format(offsetsChannel.host, offsetsChannel.port))
    }
  }

  def shutdown() {
    val canShutdown = isShuttingDown.compareAndSet(false, true)
    if (canShutdown) {
      info("ZKConsumerConnector shutting down")
      val startTime = System.nanoTime()
      KafkaMetricsGroup.removeAllConsumerMetrics(config.clientId)
      if (wildcardTopicWatcher != null)
        wildcardTopicWatcher.shutdown()
      rebalanceLock synchronized {
        try {
          if (config.autoCommitEnable)
	        scheduler.shutdown()
          fetcher match {
            case Some(f) => f.stopConnections
            case None =>
          }
          sendShutdownToAllQueues()
          if (config.autoCommitEnable)
            commitOffsets(true)
          if (zkUtils != null) {
            zkUtils.close()
            zkUtils = null
          }

          if (offsetsChannel != null) offsetsChannel.disconnect()
        } catch {
          case e: Throwable =>
            fatal("error during consumer connector shutdown", e)
        }
        info("ZKConsumerConnector shutdown completed in " + (System.nanoTime() - startTime) / 1000000 + " ms")
      }
    }
  }

  def consume[K, V](topicCountMap: scala.collection.Map[String,Int], keyDecoder: Decoder[K], valueDecoder: Decoder[V])
      : Map[String,List[KafkaStream[K,V]]] = {
    debug("entering consume ")
    if (topicCountMap == null)
      throw new RuntimeException("topicCountMap is null")

    val topicCount = TopicCount.constructTopicCount(consumerIdString, topicCountMap)

    val topicThreadIds = topicCount.getConsumerThreadIdsPerTopic

    // make a list of (queue,stream) pairs, one pair for each threadId
    val queuesAndStreams = topicThreadIds.values.map(threadIdSet =>
      threadIdSet.map(_ => {
        val queue =  new LinkedBlockingQueue[FetchedDataChunk](config.queuedMaxMessages)
        val stream = new KafkaStream[K,V](
          queue, config.consumerTimeoutMs, keyDecoder, valueDecoder, config.clientId)
        (queue, stream)
      })
    ).flatten.toList

    val dirs = new ZKGroupDirs(config.groupId)
    registerConsumerInZK(dirs, consumerIdString, topicCount)
    reinitializeConsumer(topicCount, queuesAndStreams)

    loadBalancerListener.kafkaMessageAndMetadataStreams.asInstanceOf[Map[String, List[KafkaStream[K,V]]]]
  }

  // this API is used by unit tests only
  def getTopicRegistry: Pool[String, Pool[Int, PartitionTopicInfo]] = topicRegistry

  private def registerConsumerInZK(dirs: ZKGroupDirs, consumerIdString: String, topicCount: TopicCount) {
    info("begin registering consumer " + consumerIdString + " in ZK")
    val timestamp = SystemTime.milliseconds.toString
    val consumerRegistrationInfo = Json.encode(Map("version" -> 1, "subscription" -> topicCount.getTopicCountMap, "pattern" -> topicCount.pattern,
                                                  "timestamp" -> timestamp))
    val zkWatchedEphemeral = new ZKCheckedEphemeral(dirs.
                                                    consumerRegistryDir + "/" + consumerIdString,
                                                    consumerRegistrationInfo,
                                                    zkUtils.zkConnection.getZookeeper,
                                                    false)
    zkWatchedEphemeral.create()

    info("end registering consumer " + consumerIdString + " in ZK")
  }

  private def sendShutdownToAllQueues() = {
    for (queue <- topicThreadIdAndQueues.values.toSet[BlockingQueue[FetchedDataChunk]]) {
      debug("Clearing up queue")
      queue.clear()
      queue.put(ZookeeperConsumerConnector.shutdownCommand)
      debug("Cleared queue and sent shutdown command")
    }
  }

  def autoCommit() {
    trace("auto committing")
    try {
      commitOffsets(isAutoCommit = false)
    }
    catch {
      case t: Throwable =>
      // log it and let it go
        error("exception during autoCommit: ", t)
    }
  }

  def commitOffsetToZooKeeper(topicPartition: TopicAndPartition, offset: Long) {
    if (checkpointedZkOffsets.get(topicPartition) != offset) {
      val topicDirs = new ZKGroupTopicDirs(config.groupId, topicPartition.topic)
      zkUtils.updatePersistentPath(topicDirs.consumerOffsetDir + "/" + topicPartition.partition, offset.toString)
      checkpointedZkOffsets.put(topicPartition, offset)
      zkCommitMeter.mark()
    }
  }

  /**
   * KAFKA-1743: This method added for backward compatibility.
   */
  def commitOffsets { commitOffsets(true) }

  def commitOffsets(isAutoCommit: Boolean) {

    val offsetsToCommit =
      immutable.Map(topicRegistry.values.flatMap { partitionTopicInfos =>
        partitionTopicInfos.values.map { info =>
          TopicAndPartition(info.topic, info.partitionId) -> OffsetAndMetadata(info.getConsumeOffset())
        }
      }.toSeq: _*)

    commitOffsets(offsetsToCommit, isAutoCommit)

  }

  def commitOffsets(offsetsToCommit: immutable.Map[TopicAndPartition, OffsetAndMetadata], isAutoCommit: Boolean) {
    trace("OffsetMap: %s".format(offsetsToCommit))
    var retriesRemaining = 1 + (if (isAutoCommit) 0 else config.offsetsCommitMaxRetries) // no retries for commits from auto-commit
    var done = false
    while (!done) {
      val committed = offsetsChannelLock synchronized {
        // committed when we receive either no error codes or only MetadataTooLarge errors
        if (offsetsToCommit.size > 0) {
          if (config.offsetsStorage == "zookeeper") {
            offsetsToCommit.foreach { case (topicAndPartition, offsetAndMetadata) =>
              commitOffsetToZooKeeper(topicAndPartition, offsetAndMetadata.offset)
            }
            true
          } else {
            val offsetCommitRequest = OffsetCommitRequest(config.groupId, offsetsToCommit, clientId = config.clientId)
            ensureOffsetManagerConnected()
            try {
              kafkaCommitMeter.mark(offsetsToCommit.size)
              offsetsChannel.send(offsetCommitRequest)
              val offsetCommitResponse = OffsetCommitResponse.readFrom(offsetsChannel.receive().payload())
              trace("Offset commit response: %s.".format(offsetCommitResponse))

              val (commitFailed, retryableIfFailed, shouldRefreshCoordinator, errorCount) = {
                offsetCommitResponse.commitStatus.foldLeft(false, false, false, 0) { case (folded, (topicPartition, errorCode)) =>

                  if (errorCode == Errors.NONE.code && config.dualCommitEnabled) {
                    val offset = offsetsToCommit(topicPartition).offset
                    commitOffsetToZooKeeper(topicPartition, offset)
                  }

                  (folded._1 || // update commitFailed
                    errorCode != Errors.NONE.code,

                    folded._2 || // update retryableIfFailed - (only metadata too large is not retryable)
                      (errorCode != Errors.NONE.code && errorCode != Errors.OFFSET_METADATA_TOO_LARGE.code),

                    folded._3 || // update shouldRefreshCoordinator
                      errorCode == Errors.NOT_COORDINATOR_FOR_GROUP.code ||
                      errorCode == Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code,

                    // update error count
                    folded._4 + (if (errorCode != Errors.NONE.code) 1 else 0))
                }
              }
              debug(errorCount + " errors in offset commit response.")

              if (shouldRefreshCoordinator) {
                debug("Could not commit offsets (because offset coordinator has moved or is unavailable).")
                offsetsChannel.disconnect()
              }

              if (commitFailed && retryableIfFailed)
                false
              else
                true
            }
            catch {
              case t: Throwable =>
                error("Error while committing offsets.", t)
                offsetsChannel.disconnect()
                false
            }
          }
        } else {
          debug("No updates to offsets since last commit.")
          true
        }
      }

      done = {
        retriesRemaining -= 1
        retriesRemaining == 0 || committed
      }

      if (!done) {
        debug("Retrying offset commit in %d ms".format(config.offsetsChannelBackoffMs))
        Thread.sleep(config.offsetsChannelBackoffMs)
      }
    }
  }

  private def fetchOffsetFromZooKeeper(topicPartition: TopicAndPartition) = {
    val dirs = new ZKGroupTopicDirs(config.groupId, topicPartition.topic)
    val offsetString = zkUtils.readDataMaybeNull(dirs.consumerOffsetDir + "/" + topicPartition.partition)._1
    offsetString match {
      case Some(offsetStr) => (topicPartition, OffsetMetadataAndError(offsetStr.toLong))
      case None => (topicPartition, OffsetMetadataAndError.NoOffset)
    }
  }

  private def fetchOffsets(partitions: Seq[TopicAndPartition]) = {
    if (partitions.isEmpty)
      Some(OffsetFetchResponse(Map.empty))
    else if (config.offsetsStorage == "zookeeper") {
      val offsets = partitions.map(fetchOffsetFromZooKeeper)
      Some(OffsetFetchResponse(immutable.Map(offsets:_*)))
    } else {
      val offsetFetchRequest = OffsetFetchRequest(groupId = config.groupId, requestInfo = partitions, clientId = config.clientId)

      var offsetFetchResponseOpt: Option[OffsetFetchResponse] = None
      while (!isShuttingDown.get && !offsetFetchResponseOpt.isDefined) {
        offsetFetchResponseOpt = offsetsChannelLock synchronized {
          ensureOffsetManagerConnected()
          try {
            offsetsChannel.send(offsetFetchRequest)
            val offsetFetchResponse = OffsetFetchResponse.readFrom(offsetsChannel.receive().payload())
            trace("Offset fetch response: %s.".format(offsetFetchResponse))

            val (leaderChanged, loadInProgress) =
              offsetFetchResponse.requestInfo.values.foldLeft(false, false) { case (folded, offsetMetadataAndError) =>
                (folded._1 || (offsetMetadataAndError.error == Errors.NOT_COORDINATOR_FOR_GROUP.code),
                 folded._2 || (offsetMetadataAndError.error == Errors.GROUP_LOAD_IN_PROGRESS.code))
              }

            if (leaderChanged) {
              offsetsChannel.disconnect()
              debug("Could not fetch offsets (because offset manager has moved).")
              None // retry
            }
            else if (loadInProgress) {
              debug("Could not fetch offsets (because offset cache is being loaded).")
              None // retry
            }
            else {
              if (config.dualCommitEnabled) {
                // if dual-commit is enabled (i.e., if a consumer group is migrating offsets to kafka), then pick the
                // maximum between offsets in zookeeper and kafka.
                val kafkaOffsets = offsetFetchResponse.requestInfo
                val mostRecentOffsets = kafkaOffsets.map { case (topicPartition, kafkaOffset) =>
                  val zkOffset = fetchOffsetFromZooKeeper(topicPartition)._2.offset
                  val mostRecentOffset = zkOffset.max(kafkaOffset.offset)
                  (topicPartition, OffsetMetadataAndError(mostRecentOffset, kafkaOffset.metadata, Errors.NONE.code))
                }
                Some(OffsetFetchResponse(mostRecentOffsets))
              }
              else
                Some(offsetFetchResponse)
            }
          }
          catch {
            case e: Exception =>
              warn("Error while fetching offsets from %s:%d. Possible cause: %s".format(offsetsChannel.host, offsetsChannel.port, e.getMessage))
              offsetsChannel.disconnect()
              None // retry
          }
        }

        if (offsetFetchResponseOpt.isEmpty) {
          debug("Retrying offset fetch in %d ms".format(config.offsetsChannelBackoffMs))
          Thread.sleep(config.offsetsChannelBackoffMs)
        }
      }

      offsetFetchResponseOpt
    }
  }

  class ZKSessionExpireListener(val dirs: ZKGroupDirs,
                                 val consumerIdString: String,
                                 val topicCount: TopicCount,
                                 val loadBalancerListener: ZKRebalancerListener)
    extends IZkStateListener {
    @throws(classOf[Exception])
    def handleStateChanged(state: KeeperState) {
      // do nothing, since zkclient will do reconnect for us.
    }

    /**
     * Called after the zookeeper session has expired and a new session has been created. You would have to re-create
     * any ephemeral nodes here.
     *
     * @throws Exception
     *             On any error.
     */
    @throws(classOf[Exception])
    def handleNewSession() {
      /**
       *  When we get a SessionExpired event, we lost all ephemeral nodes and zkclient has reestablished a
       *  connection for us. We need to release the ownership of the current consumer and re-register this
       *  consumer in the consumer registry and trigger a rebalance.
       */
      info("ZK expired; release old broker parition ownership; re-register consumer " + consumerIdString)
      loadBalancerListener.resetState()
      registerConsumerInZK(dirs, consumerIdString, topicCount)
      // explicitly trigger load balancing for this consumer
      loadBalancerListener.syncedRebalance()
      // There is no need to resubscribe to child and state changes.
      // The child change watchers will be set inside rebalance when we read the children list.
    }

    override def handleSessionEstablishmentError(error: Throwable): Unit = {
      fatal("Could not establish session with zookeeper", error)
    }
  }

  class ZKTopicPartitionChangeListener(val loadBalancerListener: ZKRebalancerListener)
    extends IZkDataListener {

    def handleDataChange(dataPath : String, data: Object) {
      try {
        info("Topic info for path " + dataPath + " changed to " + data.toString + ", triggering rebalance")
        // queue up the rebalance event
        loadBalancerListener.rebalanceEventTriggered()
        // There is no need to re-subscribe the watcher since it will be automatically
        // re-registered upon firing of this event by zkClient
      } catch {
        case e: Throwable => error("Error while handling topic partition change for data path " + dataPath, e )
      }
    }

    @throws(classOf[Exception])
    def handleDataDeleted(dataPath : String) {
      // TODO: This need to be implemented when we support delete topic
      warn("Topic for path " + dataPath + " gets deleted, which should not happen at this time")
    }
  }

  class ZKRebalancerListener(val group: String, val consumerIdString: String,
                             val kafkaMessageAndMetadataStreams: mutable.Map[String,List[KafkaStream[_,_]]])
    extends IZkChildListener {

    private val partitionAssignor = PartitionAssignor.createInstance(config.partitionAssignmentStrategy)

    private var isWatcherTriggered = false
    private val lock = new ReentrantLock
    private val cond = lock.newCondition()

    @volatile private var allTopicsOwnedPartitionsCount = 0
    newGauge("OwnedPartitionsCount",
      new Gauge[Int] {
        def value() = allTopicsOwnedPartitionsCount
      },
      Map("clientId" -> config.clientId, "groupId" -> config.groupId))

    private def ownedPartitionsCountMetricTags(topic: String) = Map("clientId" -> config.clientId, "groupId" -> config.groupId, "topic" -> topic)

    private val watcherExecutorThread = new Thread(consumerIdString + "_watcher_executor") {
      override def run() {
        info("starting watcher executor thread for consumer " + consumerIdString)
        var doRebalance = false
        while (!isShuttingDown.get) {
          try {
            lock.lock()
            try {
              if (!isWatcherTriggered)
                cond.await(1000, TimeUnit.MILLISECONDS) // wake up periodically so that it can check the shutdown flag
            } finally {
              doRebalance = isWatcherTriggered
              isWatcherTriggered = false
              lock.unlock()
            }
            if (doRebalance)
              syncedRebalance
          } catch {
            case t: Throwable => error("error during syncedRebalance", t)
          }
        }
        info("stopping watcher executor thread for consumer " + consumerIdString)
      }
    }
    watcherExecutorThread.start()

    @throws(classOf[Exception])
    def handleChildChange(parentPath : String, curChilds : java.util.List[String]) {
      rebalanceEventTriggered()
    }

    def rebalanceEventTriggered() {
      inLock(lock) {
        isWatcherTriggered = true
        cond.signalAll()
      }
    }

    private def deletePartitionOwnershipFromZK(topic: String, partition: Int) {
      val topicDirs = new ZKGroupTopicDirs(group, topic)
      val znode = topicDirs.consumerOwnerDir + "/" + partition
      zkUtils.deletePath(znode)
      debug("Consumer " + consumerIdString + " releasing " + znode)
    }

    private def releasePartitionOwnership(localTopicRegistry: Pool[String, Pool[Int, PartitionTopicInfo]])= {
      info("Releasing partition ownership")
      for ((topic, infos) <- localTopicRegistry) {
        for(partition <- infos.keys) {
          deletePartitionOwnershipFromZK(topic, partition)
        }
        removeMetric("OwnedPartitionsCount", ownedPartitionsCountMetricTags(topic))
        localTopicRegistry.remove(topic)
      }
      allTopicsOwnedPartitionsCount = 0
    }

    def resetState() {
      topicRegistry.clear
    }

    def syncedRebalance() {
      rebalanceLock synchronized {
        rebalanceTimer.time {
          for (i <- 0 until config.rebalanceMaxRetries) {
            if(isShuttingDown.get())  {
              return
            }
            info("begin rebalancing consumer " + consumerIdString + " try #" + i)
            var done = false
            var cluster: Cluster = null
            try {
              cluster = zkUtils.getCluster()
              done = rebalance(cluster)
            } catch {
              case e: Throwable =>
                /** occasionally, we may hit a ZK exception because the ZK state is changing while we are iterating.
                  * For example, a ZK node can disappear between the time we get all children and the time we try to get
                  * the value of a child. Just let this go since another rebalance will be triggered.
                  **/
                info("exception during rebalance ", e)
            }
            info("end rebalancing consumer " + consumerIdString + " try #" + i)
            if (done) {
              return
            } else {
              /* Here the cache is at a risk of being stale. To take future rebalancing decisions correctly, we should
               * clear the cache */
              info("Rebalancing attempt failed. Clearing the cache before the next rebalancing operation is triggered")
            }
            // stop all fetchers and clear all the queues to avoid data duplication
            closeFetchersForQueues(cluster, kafkaMessageAndMetadataStreams, topicThreadIdAndQueues.map(q => q._2))
            Thread.sleep(config.rebalanceBackoffMs)
          }
        }
      }

      throw new ConsumerRebalanceFailedException(consumerIdString + " can't rebalance after " + config.rebalanceMaxRetries +" retries")
    }

    private def rebalance(cluster: Cluster): Boolean = {
      val myTopicThreadIdsMap = TopicCount.constructTopicCount(
        group, consumerIdString, zkUtils, config.excludeInternalTopics).getConsumerThreadIdsPerTopic
      val brokers = zkUtils.getAllBrokersInCluster()
      if (brokers.size == 0) {
        // This can happen in a rare case when there are no brokers available in the cluster when the consumer is started.
        // We log an warning and register for child changes on brokers/id so that rebalance can be triggered when the brokers
        // are up.
        warn("no brokers found when trying to rebalance.")
        zkUtils.zkClient.subscribeChildChanges(BrokerIdsPath, loadBalancerListener)
        true
      }
      else {
        /**
         * fetchers must be stopped to avoid data duplication, since if the current
         * rebalancing attempt fails, the partitions that are released could be owned by another consumer.
         * But if we don't stop the fetchers first, this consumer would continue returning data for released
         * partitions in parallel. So, not stopping the fetchers leads to duplicate data.
         */
        closeFetchers(cluster, kafkaMessageAndMetadataStreams, myTopicThreadIdsMap)
        if (consumerRebalanceListener != null) {
          info("Invoking rebalance listener before relasing partition ownerships.")
          consumerRebalanceListener.beforeReleasingPartitions(
            if (topicRegistry.size == 0)
              new java.util.HashMap[String, java.util.Set[java.lang.Integer]]
            else
              topicRegistry.map(topics =>
                topics._1 -> topics._2.keys   // note this is incorrect, see KAFKA-2284
              ).toMap.asJava.asInstanceOf[java.util.Map[String, java.util.Set[java.lang.Integer]]]
          )
        }
        releasePartitionOwnership(topicRegistry)
        val assignmentContext = new AssignmentContext(group, consumerIdString, config.excludeInternalTopics, zkUtils)
        val globalPartitionAssignment = partitionAssignor.assign(assignmentContext)
        val partitionAssignment = globalPartitionAssignment.get(assignmentContext.consumerId)
        val currentTopicRegistry = new Pool[String, Pool[Int, PartitionTopicInfo]](
          valueFactory = Some((_: String) => new Pool[Int, PartitionTopicInfo]))

        // fetch current offsets for all topic-partitions
        val topicPartitions = partitionAssignment.keySet.toSeq

        val offsetFetchResponseOpt = fetchOffsets(topicPartitions)

        if (isShuttingDown.get || !offsetFetchResponseOpt.isDefined)
          false
        else {
          val offsetFetchResponse = offsetFetchResponseOpt.get
          topicPartitions.foreach(topicAndPartition => {
            val (topic, partition) = topicAndPartition.asTuple
            val offset = offsetFetchResponse.requestInfo(topicAndPartition).offset
            val threadId = partitionAssignment(topicAndPartition)
            addPartitionTopicInfo(currentTopicRegistry, partition, topic, offset, threadId)
          })

          /**
           * move the partition ownership here, since that can be used to indicate a truly successful re-balancing attempt
           * A rebalancing attempt is completed successfully only after the fetchers have been started correctly
           */
          if(reflectPartitionOwnershipDecision(partitionAssignment)) {
            allTopicsOwnedPartitionsCount = partitionAssignment.size

            partitionAssignment.view.groupBy { case (topicPartition, _) => topicPartition.topic }
                                      .foreach { case (topic, partitionThreadPairs) =>
              newGauge("OwnedPartitionsCount",
                new Gauge[Int] {
                  def value() = partitionThreadPairs.size
                },
                ownedPartitionsCountMetricTags(topic))
            }

            topicRegistry = currentTopicRegistry
            // Invoke beforeStartingFetchers callback if the consumerRebalanceListener is set.
            if (consumerRebalanceListener != null) {
              info("Invoking rebalance listener before starting fetchers.")

              // Partition assignor returns the global partition assignment organized as a map of [TopicPartition, ThreadId]
              // per consumer, and we need to re-organize it to a map of [Partition, ThreadId] per topic before passing
              // to the rebalance callback.
              val partitionAssginmentGroupByTopic = globalPartitionAssignment.values.flatten.groupBy[String] {
                case (topicPartition, _) => topicPartition.topic
              }
              val partitionAssigmentMapForCallback = partitionAssginmentGroupByTopic.map({
                case (topic, partitionOwnerShips) =>
                  val partitionOwnershipForTopicScalaMap = partitionOwnerShips.map({
                    case (topicAndPartition, consumerThreadId) =>
                      (topicAndPartition.partition: Integer) -> consumerThreadId
                  }).toMap
                  topic -> partitionOwnershipForTopicScalaMap.asJava
              })
              consumerRebalanceListener.beforeStartingFetchers(
                consumerIdString,
                partitionAssigmentMapForCallback.asJava
              )
            }
            updateFetcher(cluster)
            true
          } else {
            false
          }
        }
      }
    }

    private def closeFetchersForQueues(cluster: Cluster,
                                       messageStreams: Map[String,List[KafkaStream[_,_]]],
                                       queuesToBeCleared: Iterable[BlockingQueue[FetchedDataChunk]]) {
      val allPartitionInfos = topicRegistry.values.map(p => p.values).flatten
      fetcher match {
        case Some(f) =>
          f.stopConnections
          clearFetcherQueues(allPartitionInfos, cluster, queuesToBeCleared, messageStreams)
          /**
          * here, we need to commit offsets before stopping the consumer from returning any more messages
          * from the current data chunk. Since partition ownership is not yet released, this commit offsets
          * call will ensure that the offsets committed now will be used by the next consumer thread owning the partition
          * for the current data chunk. Since the fetchers are already shutdown and this is the last chunk to be iterated
          * by the consumer, there will be no more messages returned by this iterator until the rebalancing finishes
          * successfully and the fetchers restart to fetch more data chunks
          **/
        if (config.autoCommitEnable) {
          info("Committing all offsets after clearing the fetcher queues")
          commitOffsets(true)
        }
        case None =>
      }
    }

    private def clearFetcherQueues(topicInfos: Iterable[PartitionTopicInfo], cluster: Cluster,
                                   queuesTobeCleared: Iterable[BlockingQueue[FetchedDataChunk]],
                                   messageStreams: Map[String,List[KafkaStream[_,_]]]) {

      // Clear all but the currently iterated upon chunk in the consumer thread's queue
      queuesTobeCleared.foreach(_.clear)
      info("Cleared all relevant queues for this fetcher")

      // Also clear the currently iterated upon chunk in the consumer threads
      if(messageStreams != null)
         messageStreams.foreach(_._2.foreach(s => s.clear()))

      info("Cleared the data chunks in all the consumer message iterators")

    }

    private def closeFetchers(cluster: Cluster, messageStreams: Map[String,List[KafkaStream[_,_]]],
                              relevantTopicThreadIdsMap: Map[String, Set[ConsumerThreadId]]) {
      // only clear the fetcher queues for certain topic partitions that *might* no longer be served by this consumer
      // after this rebalancing attempt
      val queuesTobeCleared = topicThreadIdAndQueues.filter(q => relevantTopicThreadIdsMap.contains(q._1._1)).map(q => q._2)
      closeFetchersForQueues(cluster, messageStreams, queuesTobeCleared)
    }

    private def updateFetcher(cluster: Cluster) {
      // update partitions for fetcher
      var allPartitionInfos : List[PartitionTopicInfo] = Nil
      for (partitionInfos <- topicRegistry.values)
        for (partition <- partitionInfos.values)
          allPartitionInfos ::= partition
      info("Consumer " + consumerIdString + " selected partitions : " +
        allPartitionInfos.sortWith((s,t) => s.partitionId < t.partitionId).map(_.toString).mkString(","))

      fetcher match {
        case Some(f) =>
          f.startConnections(allPartitionInfos, cluster)
        case None =>
      }
    }

    private def reflectPartitionOwnershipDecision(partitionAssignment: Map[TopicAndPartition, ConsumerThreadId]): Boolean = {
      var successfullyOwnedPartitions : List[(String, Int)] = Nil
      val partitionOwnershipSuccessful = partitionAssignment.map { partitionOwner =>
        val topic = partitionOwner._1.topic
        val partition = partitionOwner._1.partition
        val consumerThreadId = partitionOwner._2
        val partitionOwnerPath = zkUtils.getConsumerPartitionOwnerPath(group, topic, partition)
        try {
          zkUtils.createEphemeralPathExpectConflict(partitionOwnerPath, consumerThreadId.toString)
          info(consumerThreadId + " successfully owned partition " + partition + " for topic " + topic)
          successfullyOwnedPartitions ::= (topic, partition)
          true
        } catch {
          case _: ZkNodeExistsException =>
            // The node hasn't been deleted by the original owner. So wait a bit and retry.
            info("waiting for the partition ownership to be deleted: " + partition)
            false
        }
      }
      val hasPartitionOwnershipFailed = partitionOwnershipSuccessful.foldLeft(0)((sum, decision) => sum + (if(decision) 0 else 1))
      /* even if one of the partition ownership attempt has failed, return false */
      if(hasPartitionOwnershipFailed > 0) {
        // remove all paths that we have owned in ZK
        successfullyOwnedPartitions.foreach(topicAndPartition => deletePartitionOwnershipFromZK(topicAndPartition._1, topicAndPartition._2))
        false
      }
      else true
    }

    private def addPartitionTopicInfo(currentTopicRegistry: Pool[String, Pool[Int, PartitionTopicInfo]],
                                      partition: Int, topic: String,
                                      offset: Long, consumerThreadId: ConsumerThreadId) {
      val partTopicInfoMap = currentTopicRegistry.getAndMaybePut(topic)

      val queue = topicThreadIdAndQueues.get((topic, consumerThreadId))
      val consumedOffset = new AtomicLong(offset)
      val fetchedOffset = new AtomicLong(offset)
      val partTopicInfo = new PartitionTopicInfo(topic,
                                                 partition,
                                                 queue,
                                                 consumedOffset,
                                                 fetchedOffset,
                                                 new AtomicInteger(config.fetchMessageMaxBytes),
                                                 config.clientId)
      partTopicInfoMap.put(partition, partTopicInfo)
      debug(partTopicInfo + " selected new offset " + offset)
      checkpointedZkOffsets.put(TopicAndPartition(topic, partition), offset)
    }
  }

  private def reinitializeConsumer[K,V](
      topicCount: TopicCount,
      queuesAndStreams: List[(LinkedBlockingQueue[FetchedDataChunk],KafkaStream[K,V])]) {

    val dirs = new ZKGroupDirs(config.groupId)

    // listener to consumer and partition changes
    if (loadBalancerListener == null) {
      val topicStreamsMap = new mutable.HashMap[String,List[KafkaStream[K,V]]]
      loadBalancerListener = new ZKRebalancerListener(
        config.groupId, consumerIdString, topicStreamsMap.asInstanceOf[scala.collection.mutable.Map[String, List[KafkaStream[_,_]]]])
    }

    // create listener for session expired event if not exist yet
    if (sessionExpirationListener == null)
      sessionExpirationListener = new ZKSessionExpireListener(
        dirs, consumerIdString, topicCount, loadBalancerListener)

    // create listener for topic partition change event if not exist yet
    if (topicPartitionChangeListener == null)
      topicPartitionChangeListener = new ZKTopicPartitionChangeListener(loadBalancerListener)

    val topicStreamsMap = loadBalancerListener.kafkaMessageAndMetadataStreams

    // map of {topic -> Set(thread-1, thread-2, ...)}
    val consumerThreadIdsPerTopic: Map[String, Set[ConsumerThreadId]] =
      topicCount.getConsumerThreadIdsPerTopic

    val allQueuesAndStreams = topicCount match {
      case _: WildcardTopicCount =>
        /*
         * Wild-card consumption streams share the same queues, so we need to
         * duplicate the list for the subsequent zip operation.
         */
        (1 to consumerThreadIdsPerTopic.keySet.size).flatMap(_ => queuesAndStreams).toList
      case _: StaticTopicCount =>
        queuesAndStreams
    }

    val topicThreadIds = consumerThreadIdsPerTopic.map { case (topic, threadIds) =>
      threadIds.map((topic, _))
    }.flatten

    require(topicThreadIds.size == allQueuesAndStreams.size,
      "Mismatch between thread ID count (%d) and queue count (%d)"
      .format(topicThreadIds.size, allQueuesAndStreams.size))
    val threadQueueStreamPairs = topicThreadIds.zip(allQueuesAndStreams)

    threadQueueStreamPairs.foreach(e => {
      val topicThreadId = e._1
      val q = e._2._1
      topicThreadIdAndQueues.put(topicThreadId, q)
      debug("Adding topicThreadId %s and queue %s to topicThreadIdAndQueues data structure".format(topicThreadId, q.toString))
      newGauge(
        "FetchQueueSize",
        new Gauge[Int] {
          def value = q.size
        },
        Map("clientId" -> config.clientId,
          "topic" -> topicThreadId._1,
          "threadId" -> topicThreadId._2.threadId.toString)
      )
    })

    val groupedByTopic = threadQueueStreamPairs.groupBy(_._1._1)
    groupedByTopic.foreach(e => {
      val topic = e._1
      val streams = e._2.map(_._2._2).toList
      topicStreamsMap += (topic -> streams)
      debug("adding topic %s and %d streams to map.".format(topic, streams.size))
    })

    // listener to consumer and partition changes
    zkUtils.zkClient.subscribeStateChanges(sessionExpirationListener)

    zkUtils.zkClient.subscribeChildChanges(dirs.consumerRegistryDir, loadBalancerListener)

    topicStreamsMap.foreach { topicAndStreams =>
      // register on broker partition path changes
      val topicPath = BrokerTopicsPath + "/" + topicAndStreams._1
      zkUtils.zkClient.subscribeDataChanges(topicPath, topicPartitionChangeListener)
    }

    // explicitly trigger load balancing for this consumer
    loadBalancerListener.syncedRebalance()
  }

  class WildcardStreamsHandler[K,V](topicFilter: TopicFilter,
                                  numStreams: Int,
                                  keyDecoder: Decoder[K],
                                  valueDecoder: Decoder[V])
                                extends TopicEventHandler[String] {

    if (messageStreamCreated.getAndSet(true))
      throw new RuntimeException("Each consumer connector can create " +
        "message streams by filter at most once.")

    private val wildcardQueuesAndStreams = (1 to numStreams)
      .map(_ => {
        val queue = new LinkedBlockingQueue[FetchedDataChunk](config.queuedMaxMessages)
        val stream = new KafkaStream[K,V](queue,
                                          config.consumerTimeoutMs,
                                          keyDecoder,
                                          valueDecoder,
                                          config.clientId)
        (queue, stream)
    }).toList

     // bootstrap with existing topics
    private var wildcardTopics =
      zkUtils.getChildrenParentMayNotExist(BrokerTopicsPath)
        .filter(topic => topicFilter.isTopicAllowed(topic, config.excludeInternalTopics))

    private val wildcardTopicCount = TopicCount.constructTopicCount(
      consumerIdString, topicFilter, numStreams, zkUtils, config.excludeInternalTopics)

    val dirs = new ZKGroupDirs(config.groupId)
    registerConsumerInZK(dirs, consumerIdString, wildcardTopicCount)
    reinitializeConsumer(wildcardTopicCount, wildcardQueuesAndStreams)

    /*
     * Topic events will trigger subsequent synced rebalances.
     */
    info("Creating topic event watcher for topics " + topicFilter)
    wildcardTopicWatcher = new ZookeeperTopicEventWatcher(zkUtils, this)

    def handleTopicEvent(allTopics: Seq[String]) {
      debug("Handling topic event")

      val updatedTopics = allTopics.filter(topic => topicFilter.isTopicAllowed(topic, config.excludeInternalTopics))

      val addedTopics = updatedTopics filterNot (wildcardTopics contains)
      if (addedTopics.nonEmpty)
        info("Topic event: added topics = %s"
                             .format(addedTopics))

      /*
       * TODO: Deleted topics are interesting (and will not be a concern until
       * 0.8 release). We may need to remove these topics from the rebalance
       * listener's map in reinitializeConsumer.
       */
      val deletedTopics = wildcardTopics filterNot (updatedTopics contains)
      if (deletedTopics.nonEmpty)
        info("Topic event: deleted topics = %s"
                             .format(deletedTopics))

      wildcardTopics = updatedTopics
      info("Topics to consume = %s".format(wildcardTopics))

      if (addedTopics.nonEmpty || deletedTopics.nonEmpty)
        reinitializeConsumer(wildcardTopicCount, wildcardQueuesAndStreams)
    }

    def streams: Seq[KafkaStream[K,V]] =
      wildcardQueuesAndStreams.map(_._2)
  }
}