text:
#import "MDCInkView.h"

#import "MaterialMath.h"
#import "private/MDCInkLayer.h"
#import "private/MDCLegacyInkLayer.h"

static NSString *const MDCInkViewAnimationDelegateKey = @"MDCInkViewAnimationDelegateKey";
static NSString *const MDCInkViewInkStyleKey = @"MDCInkViewInkStyleKey";
static NSString *const MDCInkViewUsesLegacyInkRippleKey = @"MDCInkViewUsesLegacyInkRippleKey";
static NSString *const MDCInkViewMaskLayerKey = @"MDCInkViewMaskLayerKey";
static NSString *const MDCInkViewUsesCustomInkCenterKey = @"MDCInkViewUsesCustomInkCenterKey";
static NSString *const MDCInkViewCustomInkCenterKey = @"MDCInkViewCustomInkCenterKey";
static NSString *const MDCInkViewInkColorKey = @"MDCInkViewInkColorKey";
static NSString *const MDCInkViewMaxRippleRadiusKey = @"MDCInkViewMaxRippleRadiusKey";

@interface MDCInkPendingAnimation : NSObject <CAAction>

@property(nonatomic, weak) CALayer *animationSourceLayer;
@property(nonatomic, strong) NSString *keyPath;
@property(nonatomic, strong) id fromValue;
@property(nonatomic, strong) id toValue;

@end

@interface MDCInkView () <CALayerDelegate, MDCInkLayerDelegate>

@property(nonatomic, strong) CAShapeLayer *maskLayer;
@property(nonatomic, copy) MDCInkCompletionBlock startInkRippleCompletionBlock;
@property(nonatomic, copy) MDCInkCompletionBlock endInkRippleCompletionBlock;
@property(nonatomic, strong) MDCInkLayer *activeInkLayer;

// Legacy ink ripple
@property(nonatomic, readonly) MDCLegacyInkLayer *inkLayer;

@end

@implementation MDCInkView {
  CGFloat _maxRippleRadius;
}

+ (Class)layerClass {
  return [MDCLegacyInkLayer class];
}

- (instancetype)initWithFrame:(CGRect)frame {
  self = [super initWithFrame:frame];
  if (self) {
    [self commonMDCInkViewInit];
  }
  return self;
}

- (instancetype)initWithCoder:(NSCoder *)aDecoder {
  self = [super initWithCoder:aDecoder];
  if (self) {
    if ([aDecoder containsValueForKey:MDCInkViewAnimationDelegateKey]) {
      _animationDelegate = [aDecoder decodeObjectForKey:MDCInkViewAnimationDelegateKey];
    }
    if ([aDecoder containsValueForKey:MDCInkViewMaskLayerKey]) {
      _maskLayer = [aDecoder decodeObjectForKey:MDCInkViewMaskLayerKey];
      _maskLayer.delegate = self;
    } else {
      _maskLayer = [CAShapeLayer layer];
      _maskLayer.delegate = self;
    }
    if ([aDecoder containsValueForKey:MDCInkViewUsesLegacyInkRippleKey]) {
      _usesLegacyInkRipple = [aDecoder decodeBoolForKey:MDCInkViewUsesLegacyInkRippleKey];
    } else {
      _usesLegacyInkRipple = YES;
    }
    if ([aDecoder containsValueForKey:MDCInkViewInkStyleKey]) {
      self.inkStyle = [aDecoder decodeIntegerForKey:MDCInkViewInkStyleKey];
    }

    // The following are derived properties, but `layer` may not have been encoded
    if ([aDecoder containsValueForKey:MDCInkViewUsesCustomInkCenterKey]) {
      self.usesCustomInkCenter = [aDecoder decodeBoolForKey:MDCInkViewUsesCustomInkCenterKey];
    }
    if ([aDecoder containsValueForKey:MDCInkViewCustomInkCenterKey]) {
      self.customInkCenter = [aDecoder decodeCGPointForKey:MDCInkViewCustomInkCenterKey];
    }
    if ([aDecoder containsValueForKey:MDCInkViewMaxRippleRadiusKey]) {
      self.maxRippleRadius = (CGFloat)[aDecoder decodeDoubleForKey:MDCInkViewMaxRippleRadiusKey];
    }
    if ([aDecoder containsValueForKey:MDCInkViewInkColorKey]) {
      self.inkColor = [aDecoder decodeObjectOfClass:[UIColor class] forKey:MDCInkViewInkColorKey];
    }
  }
  return self;
}

- (void)encodeWithCoder:(NSCoder *)aCoder {
  [super encodeWithCoder:aCoder];

  if (self.animationDelegate && [self.animationDelegate conformsToProtocol:@protocol(NSCoding)]) {
    [aCoder encodeObject:self.animationDelegate forKey:MDCInkViewAnimationDelegateKey];
  }
  [aCoder encodeInteger:self.inkStyle forKey:MDCInkViewInkStyleKey];
  [aCoder encodeBool:self.usesLegacyInkRipple forKey:MDCInkViewUsesLegacyInkRippleKey];
  [aCoder encodeObject:self.maskLayer forKey:MDCInkViewMaskLayerKey];

  // The following are derived properties, but `layer` may not get encoded by the superclass
  [aCoder encodeBool:self.usesCustomInkCenter forKey:MDCInkViewUsesCustomInkCenterKey];
  [aCoder encodeCGPoint:self.customInkCenter forKey:MDCInkViewCustomInkCenterKey];
  [aCoder encodeDouble:self.maxRippleRadius forKey:MDCInkViewMaxRippleRadiusKey];
  [aCoder encodeObject:self.inkColor forKey:MDCInkViewInkColorKey];
}

- (void)commonMDCInkViewInit {
  self.userInteractionEnabled = NO;
  self.backgroundColor = [UIColor clearColor];
  self.autoresizingMask = UIViewAutoresizingFlexibleHeight | UIViewAutoresizingFlexibleWidth;
  self.layer.delegate = self;
  self.inkColor = self.defaultInkColor;
  _usesLegacyInkRipple = YES;

  // Use mask layer when the superview has a shadowPath.
  _maskLayer = [CAShapeLayer layer];
  _maskLayer.delegate = self;
}

- (void)layoutSubviews {
  [super layoutSubviews];

  // If the superview has a shadowPath make sure ink does not spread outside of the shadowPath.
  if (self.superview.layer.shadowPath) {
    self.maskLayer.path = self.superview.layer.shadowPath;
    self.layer.mask = _maskLayer;
  }

  CGRect inkBounds = CGRectMake(0, 0, CGRectGetWidth(self.bounds), CGRectGetHeight(self.bounds));
  self.layer.bounds = inkBounds;

  // When bounds change ensure all ink layer bounds are changed too.
  for (CALayer *layer in self.layer.sublayers) {
    if ([layer isKindOfClass:[MDCInkLayer class]]) {
      MDCInkLayer *inkLayer = (MDCInkLayer *)layer;
      inkLayer.bounds = inkBounds;
    }
  }
}

- (void)setInkStyle:(MDCInkStyle)inkStyle {
  _inkStyle = inkStyle;
  if (self.usesLegacyInkRipple) {
    switch (inkStyle) {
      case MDCInkStyleBounded:
        self.inkLayer.masksToBounds = YES;
        self.inkLayer.bounded = YES;
        break;
      case MDCInkStyleUnbounded:
        self.inkLayer.masksToBounds = NO;
        self.inkLayer.bounded = NO;
        break;
    }
  } else {
    switch(inkStyle) {
      case MDCInkStyleBounded:
        self.inkLayer.maxRippleRadius = 0;
        break;
      case MDCInkStyleUnbounded:
        self.inkLayer.maxRippleRadius = _maxRippleRadius;
        break;
    }
  }
}

- (void)setInkColor:(UIColor *)inkColor {
  if (inkColor == nil) {
    return;
  }
  self.inkLayer.inkColor = inkColor;
}

- (UIColor *)inkColor {
  return self.inkLayer.inkColor;
}

- (CGFloat)maxRippleRadius {
  return self.inkLayer.maxRippleRadius;
}

- (void)setMaxRippleRadius:(CGFloat)radius {
  // Keep track of the set value in case the caller will change inkStyle later
  _maxRippleRadius = radius;
  if (MDCCGFloatEqual(self.inkLayer.maxRippleRadius, radius)) {
    return;
  }

  // Legacy Ink updates inkLayer.maxRippleRadius regardless of inkStyle
  if (self.usesLegacyInkRipple) {
    self.inkLayer.maxRippleRadius = radius;
    // This is required for legacy Ink so that the Ink bounds will be adjusted correctly
    [self setNeedsLayout];
  } else {
    // New Ink Bounded style ignores maxRippleRadius
    switch(self.inkStyle) {
      case MDCInkStyleUnbounded:
        self.inkLayer.maxRippleRadius = radius;
        break;
      case MDCInkStyleBounded:
        // No-op
        break;
    }
  }
}

- (BOOL)usesCustomInkCenter {
  return self.inkLayer.useCustomInkCenter;
}

- (void)setUsesCustomInkCenter:(BOOL)usesCustomInkCenter {
  self.inkLayer.useCustomInkCenter = usesCustomInkCenter;
}

- (CGPoint)customInkCenter {
  return self.inkLayer.customInkCenter;
}

- (void)setCustomInkCenter:(CGPoint)customInkCenter {
  self.inkLayer.customInkCenter = customInkCenter;
}

- (MDCLegacyInkLayer *)inkLayer {
  return (MDCLegacyInkLayer *)self.layer;
}

- (void)startTouchBeganAnimationAtPoint:(CGPoint)point
                             completion:(MDCInkCompletionBlock)completionBlock {
  [self startTouchBeganAtPoint:point animated:YES withCompletion:completionBlock];
}

- (void)startTouchBeganAtPoint:(CGPoint)point animated:(BOOL)animated
                withCompletion:(nullable MDCInkCompletionBlock)completionBlock {
  if (self.usesLegacyInkRipple) {
    [self.inkLayer spreadFromPoint:point completion:completionBlock];
  } else {
    self.startInkRippleCompletionBlock = completionBlock;
    MDCInkLayer *inkLayer = [MDCInkLayer layer];
    inkLayer.inkColor = self.inkColor;
    inkLayer.maxRippleRadius = self.maxRippleRadius;
    inkLayer.animationDelegate = self;
    inkLayer.opacity = 0;
    inkLayer.frame = self.bounds;
    [self.layer addSublayer:inkLayer];
    [inkLayer startInkAtPoint:point animated:animated];
    self.activeInkLayer = inkLayer;
  }
}

- (void)startTouchEndAtPoint:(CGPoint)point animated:(BOOL)animated
              withCompletion:(nullable MDCInkCompletionBlock)completionBlock {
  if (self.usesLegacyInkRipple) {
    [self.inkLayer evaporateWithCompletion:completionBlock];
  } else {
    self.endInkRippleCompletionBlock = completionBlock;
    [self.activeInkLayer endInkAtPoint:point animated:animated];
  }
}

- (void)startTouchEndedAnimationAtPoint:(CGPoint)point
                             completion:(MDCInkCompletionBlock)completionBlock {
  [self startTouchEndAtPoint:point animated:YES withCompletion:completionBlock];
}

- (void)cancelAllAnimationsAnimated:(BOOL)animated {
  if (self.usesLegacyInkRipple) {
    [self.inkLayer resetAllInk:animated];
  } else {
    NSArray<CALayer *> *sublayers = [self.layer.sublayers copy];
    for (CALayer *layer in sublayers) {
      if ([layer isKindOfClass:[MDCInkLayer class]]) {
        MDCInkLayer *inkLayer = (MDCInkLayer *)layer;
        if (animated) {
          [inkLayer endAnimationAtPoint:CGPointZero];
        } else {
          [inkLayer removeFromSuperlayer];
        }
      }
    }
  }
}

- (UIColor *)defaultInkColor {
  return [[UIColor alloc] initWithWhite:0 alpha:0.14f];
}

+ (MDCInkView *)injectedInkViewForView:(UIView *)view {
  MDCInkView *foundInkView = nil;
  for (MDCInkView *subview in view.subviews) {
    if ([subview isKindOfClass:[MDCInkView class]]) {
      foundInkView = subview;
      break;
    }
  }
  if (!foundInkView) {
    foundInkView = [[MDCInkView alloc] initWithFrame:view.bounds];
    foundInkView.autoresizingMask =
        UIViewAutoresizingFlexibleWidth | UIViewAutoresizingFlexibleHeight;
    [view addSubview:foundInkView];
  }
  return foundInkView;
}

#pragma mark - MDCInkLayerDelegate

- (void)inkLayerAnimationDidStart:(MDCInkLayer *)inkLayer {
  if (self.activeInkLayer == inkLayer && self.startInkRippleCompletionBlock) {
    self.startInkRippleCompletionBlock();
  }
  if ([self.animationDelegate respondsToSelector:@selector(inkAnimationDidStart:)]) {
    [self.animationDelegate inkAnimationDidStart:self];
  }
}

- (void)inkLayerAnimationDidEnd:(MDCInkLayer *)inkLayer {
  if (self.activeInkLayer == inkLayer && self.endInkRippleCompletionBlock) {
    self.endInkRippleCompletionBlock();
  }
  if ([self.animationDelegate respondsToSelector:@selector(inkAnimationDidEnd:)]) {
    [self.animationDelegate inkAnimationDidEnd:self];
  }
}

#pragma mark - CALayerDelegate

- (id<CAAction>)actionForLayer:(CALayer *)layer forKey:(NSString *)event {
  if ([event isEqualToString:@"path"] || [event isEqualToString:@"shadowPath"]) {

    // We have to create a pending animation because if we are inside a UIKit animation block we
    // won't know any properties of the animation block until it is commited.
    MDCInkPendingAnimation *pendingAnim = [[MDCInkPendingAnimation alloc] init];
    pendingAnim.animationSourceLayer = self.superview.layer;
    pendingAnim.fromValue = [layer.presentationLayer valueForKey:event];
    pendingAnim.toValue = nil;
    pendingAnim.keyPath = event;

    return pendingAnim;
  }
  return nil;
}

@end

@implementation MDCInkPendingAnimation

- (void)runActionForKey:(NSString *)event object:(id)anObject arguments:(NSDictionary *)dict {
  if ([anObject isKindOfClass:[CAShapeLayer class]]) {
    CAShapeLayer *layer = (CAShapeLayer *)anObject;

    // In order to synchronize our animation with UIKit animations we have to fetch the resizing
    // animation created by UIKit and copy the configuration to our custom animation.
    CAAnimation *boundsAction = [self.animationSourceLayer animationForKey:@"bounds.size"];
    if ([boundsAction isKindOfClass:[CABasicAnimation class]]) {
      CABasicAnimation *animation = (CABasicAnimation *)[boundsAction copy];
      animation.keyPath = self.keyPath;
      animation.fromValue = self.fromValue;
      animation.toValue = self.toValue;

      [layer addAnimation:animation forKey:event];
    }
  }
}

@end


text:
define(function () {

    var DoCreateError = function ( errorCode, errorMsg, errorType, errorSource ) {

        var errorObj = {
            errorCode: errorCode,
            errorMsg: errorMsg,
            errorType: errorType,
            errorSource: errorSource
        };

        return errorObj;
    };

    return {
        CreateError: DoCreateError
    }

});

redpajama_set_name: RedPajamaGithub


text:
<?php

namespace Lechimp\STG\Closures;

use Lechimp\STG\STG;
use Lechimp\STG\GC;
use Lechimp\STG\Exceptions\BlackHole;
use Lechimp\STG\CodeLabel;

/**
 * Representation of a closure in PHP.
 */
abstract class Standard extends Closure
{
    use GC;

    /**
     * @var array|null $name => $content
     */
    protected $free_variables;

    // For updating and garbage collection.
    /**
     * @var Closure|null
     */
    public $updated;

    /**
     * ATTENTION: The dictionary of free variables is passed by reference.
     *            to make recursive definitions possible.
     */
    public function __construct(array &$free_variables)
    {
        $free_variables_names = $this->free_variables_names();
        assert(count($free_variables_names) == count($free_variables));
        assert(sort($free_variables_names) == sort(array_keys($free_variables)));
        $this->free_variables = &$free_variables;
        $this->entry_code = new CodeLabel($this, "entry_code");
        $this->updated = null;

        // For partial update.
        $this->function_closure = null;
        $this->argument_stack = null;
        $this->return_stack = null;
        $this->env_stack = null;
    }

    /**
     * Get a list of the free variables of the closure.
     *
     * @return  string[]
     */
    abstract public function free_variables_names();

    /**
     * Black hole entry code.
     *
     * This is used entry code when this closure is just evaluated.
     *
     * @param   STG $stg
     */
    public function black_hole(STG $stg)
    {
        throw new BlackHole();
    }

    /**
     * Get a free variable.
     *
     * @param   string          $name
     * @return  Closures\Closure|int
     */
    public function free_variable($name)
    {
        assert(is_string($name));
        assert($this->updated === null);
        if (array_key_exists($name, $this->free_variables)) {
            throw new \LogicException("Unknown free variable '$name' in closure '"
                                     . get_class($this) . "'");
        }
        return $this->free_variables[$name];
    }

    /**
     * Update this closure with another closure.
     *
     * Cleans free variables, sets pointer to updated version of this closure
     * and overwrites entry code of this closure to the updated version.
     *
     * @param   Closure $updated
     * @return  null
     */
    public function update(Closure $updated)
    {
        assert($this->updated === null);
        $this->updated = $updated;
        $this->free_variables = null;
        $this->entry_code = $updated->entry_code;
    }

    /**
     * @inheritdoc
     */
    public function collect_garbage(array &$survivors)
    {
        if ($this->updated !== null) {
            // The closure was updated. Drop it.
            return $this->updated->collect_garbage($survivors);
        }

        return parent::collect_garbage($survivors);
    }

    /**
     * @inheritdoc
     */
    public function collect_garbage_in_references(array &$survivors)
    {
        assert($this->free_variables !== null);
        $this->collect_garbage_in_array($this->free_variables, $survivors);
    }
}


text:
package org.apache.kafka.clients.consumer;

import org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.protocol.types.ArrayOf;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.protocol.types.Schema;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.protocol.types.Type;
import org.apache.kafka.common.utils.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeSet;

/**
 *
The sticky assignor serves two purposes. First, it guarantees an assignment that is as balanced as possible, meaning either:
 *

 *
the numbers of topic partitions assigned to consumers differ by at most one; or

 *
each consumer that has 2+ fewer topic partitions than some other consumer cannot get any of those topic partitions transferred to it.

 *

 * Second, it preserved as many existing assignment as possible when a reassignment occurs. This helps in saving some of the
 * overhead processing when topic partitions move from one consumer to another.

 *
 *
Starting fresh it would work by distributing the partitions over consumers as evenly as possible. Even though this may sound similar to
 * how round robin assignor works, the second example below shows that it is not.
 * During a reassignment it would perform the reassignment in such a way that in the new assignment
 *

 *
topic partitions are still distributed as evenly as possible, and

 *
topic partitions stay with their previously assigned consumers as much as possible.

 *

 * Of course, the first goal above takes precedence over the second one.

 *
 *
Example 1. Suppose there are three consumers C0, C1, C2,
 * four topics t0, t1, t2, t3, and each topic has 2 partitions,
 * resulting in partitions t0p0, t0p1, t1p0, t1p1, t2p0,
 * t2p1, t3p0, t3p1. Each consumer is subscribed to all three topics.
 *
 * The assignment with both sticky and round robin assignors will be:
 *

 *
C0: [t0p0, t1p1, t3p0]

 *
C1: [t0p1, t2p0, t3p1]

 *
C2: [t1p0, t2p1]

 *

 *
 * Now, let's assume C1 is removed and a reassignment is about to happen. The round robin assignor would produce:
 *

 *
C0: [t0p0, t1p0, t2p0, t3p0]

 *
C2: [t0p1, t1p1, t2p1, t3p1]

 *

 *
 * while the sticky assignor would result in:
 *

 *
C0 [t0p0, t1p1, t3p0, t2p0]

 *
C2 [t1p0, t2p1, t0p1, t3p1]

 *

 * preserving all the previous assignments (unlike the round robin assignor).
 *

 *
Example 2. There are three consumers C0, C1, C2,
 * and three topics t0, t1, t2, with 1, 2, and 3 partitions respectively.
 * Therefore, the partitions are t0p0, t1p0, t1p1, t2p0,
 * t2p1, t2p2. C0 is subscribed to t0; C1 is subscribed to
 * t0, t1; and C2 is subscribed to t0, t1, t2.
 *
 * The round robin assignor would come up with the following assignment:
 *

 *
C0 [t0p0]

 *
C1 [t1p0]

 *
C2 [t1p1, t2p0, t2p1, t2p2]

 *

 *
 * which is not as balanced as the assignment suggested by sticky assignor:
 *

 *
C0 [t0p0]

 *
C1 [t1p0, t1p1]

 *
C2 [t2p0, t2p1, t2p2]

 *

 *
 * Now, if consumer C0 is removed, these two assignors would produce the following assignments.
 * Round Robin (preserves 3 partition assignments):
 *

 *
C1 [t0p0, t1p1]

 *
C2 [t1p0, t2p0, t2p1, t2p2]

 *

 *
 * Sticky (preserves 5 partition assignments):
 *

 *
C1 [t1p0, t1p1, t0p0]

 *
C2 [t2p0, t2p1, t2p2]

 *

 *

 *
Impact on ConsumerRebalanceListener

 * The sticky assignment strategy can provide some optimization to those consumers that have some partition cleanup code
 * in their onPartitionsRevoked() callback listeners. The cleanup code is placed in that callback listener
 * because the consumer has no assumption or hope of preserving any of its assigned partitions after a rebalance when it
 * is using range or round robin assignor. The listener code would look like this:
 *

 * {@code
 * class TheOldRebalanceListener implements ConsumerRebalanceListener {
 *
 *   void onPartitionsRevoked(Collection<TopicPartition> partitions) {
 *     for (TopicPartition partition: partitions) {
 *       commitOffsets(partition);
 *       cleanupState(partition);
 *     }
 *   }
 *
 *   void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 *     for (TopicPartition partition: partitions) {
 *       initializeState(partition);
 *       initializeOffset(partition);
 *     }
 *   }
 * }
 * }
 *

 *
 * As mentioned above, one advantage of the sticky assignor is that, in general, it reduces the number of partitions that
 * actually move from one consumer to another during a reassignment. Therefore, it allows consumers to do their cleanup
 * more efficiently. Of course, they still can perform the partition cleanup in the onPartitionsRevoked()
 * listener, but they can be more efficient and make a note of their partitions before and after the rebalance, and do the
 * cleanup after the rebalance only on the partitions they have lost (which is normally not a lot). The code snippet below
 * clarifies this point:
 *

 * {@code
 * class TheNewRebalanceListener implements ConsumerRebalanceListener {
 *   Collection<TopicPartition> lastAssignment = Collections.emptyList();
 *
 *   void onPartitionsRevoked(Collection<TopicPartition> partitions) {
 *     for (TopicPartition partition: partitions)
 *       commitOffsets(partition);
 *   }
 *
 *   void onPartitionsAssigned(Collection<TopicPartition> assignment) {
 *     for (TopicPartition partition: difference(lastAssignment, assignment))
 *       cleanupState(partition);
 *
 *     for (TopicPartition partition: difference(assignment, lastAssignment))
 *       initializeState(partition);
 *
 *     for (TopicPartition partition: assignment)
 *       initializeOffset(partition);
 *
 *     this.lastAssignment = assignment;
 *   }
 * }
 * }
 *

 *
 * Any consumer that uses sticky assignment can leverage this listener like this:
 * consumer.subscribe(topics, new TheNewRebalanceListener());
 *
 */
public class StickyAssignor extends AbstractPartitionAssignor {
    private static final Logger log = LoggerFactory.getLogger(StickyAssignor.class);

    // these schemas are used for preserving consumer's previously assigned partitions
    // list and sending it as user data to the leader during a rebalance
    private static final String TOPIC_PARTITIONS_KEY_NAME = "previous_assignment";
    private static final String TOPIC_KEY_NAME = "topic";
    private static final String PARTITIONS_KEY_NAME = "partitions";
    private static final Schema TOPIC_ASSIGNMENT = new Schema(
            new Field(TOPIC_KEY_NAME, Type.STRING),
            new Field(PARTITIONS_KEY_NAME, new ArrayOf(Type.INT32)));
    private static final Schema STICKY_ASSIGNOR_USER_DATA = new Schema(
            new Field(TOPIC_PARTITIONS_KEY_NAME, new ArrayOf(TOPIC_ASSIGNMENT)));

    private List<TopicPartition> memberAssignment = null;
    private PartitionMovements partitionMovements;

    public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic,
                                                    Map<String, Subscription> subscriptions) {
        Map<String, List<TopicPartition>> currentAssignment = new HashMap<>();
        partitionMovements = new PartitionMovements();

        prepopulateCurrentAssignments(subscriptions, currentAssignment);
        boolean isFreshAssignment = currentAssignment.isEmpty();

        // a mapping of all topic partitions to all consumers that can be assigned to them
        final Map<TopicPartition, List<String>> partition2AllPotentialConsumers = new HashMap<>();
        // a mapping of all consumers to all potential topic partitions that can be assigned to them
        final Map<String, List<TopicPartition>> consumer2AllPotentialPartitions = new HashMap<>();

        // initialize partition2AllPotentialConsumers and consumer2AllPotentialPartitions in the following two for loops
        for (Entry<String, Integer> entry: partitionsPerTopic.entrySet()) {
            for (int i = 0; i < entry.getValue(); ++i)
                partition2AllPotentialConsumers.put(new TopicPartition(entry.getKey(), i), new ArrayList<String>());
        }

        for (Entry<String, Subscription> entry: subscriptions.entrySet()) {
            String consumer = entry.getKey();
            consumer2AllPotentialPartitions.put(consumer, new ArrayList<TopicPartition>());
            for (String topic: entry.getValue().topics()) {
                for (int i = 0; i < partitionsPerTopic.get(topic); ++i) {
                    TopicPartition topicPartition = new TopicPartition(topic, i);
                    consumer2AllPotentialPartitions.get(consumer).add(topicPartition);
                    partition2AllPotentialConsumers.get(topicPartition).add(consumer);
                }
            }

            // add this consumer to currentAssignment (with an empty topic partition assignment) if it does not already exist
            if (!currentAssignment.containsKey(consumer))
                currentAssignment.put(consumer, new ArrayList<TopicPartition>());
        }

        // a mapping of partition to current consumer
        Map<TopicPartition, String> currentPartitionConsumer = new HashMap<>();
        for (Map.Entry<String, List<TopicPartition>> entry: currentAssignment.entrySet())
            for (TopicPartition topicPartition: entry.getValue())
                currentPartitionConsumer.put(topicPartition, entry.getKey());

        List<TopicPartition> sortedPartitions = sortPartitions(
                currentAssignment, isFreshAssignment, partition2AllPotentialConsumers, consumer2AllPotentialPartitions);

        // all partitions that need to be assigned (initially set to all partitions but adjusted in the following loop)
        List<TopicPartition> unassignedPartitions = new ArrayList<>(sortedPartitions);
        for (Iterator<Map.Entry<String, List<TopicPartition>>> it = currentAssignment.entrySet().iterator(); it.hasNext();) {
            Map.Entry<String, List<TopicPartition>> entry = it.next();
            if (!subscriptions.containsKey(entry.getKey())) {
                // if a consumer that existed before (and had some partition assignments) is now removed, remove it from currentAssignment
                for (TopicPartition topicPartition: entry.getValue())
                    currentPartitionConsumer.remove(topicPartition);
                it.remove();
            } else {
                // otherwise (the consumer still exists)
                for (Iterator<TopicPartition> partitionIter = entry.getValue().iterator(); partitionIter.hasNext();) {
                    TopicPartition partition = partitionIter.next();
                    if (!partition2AllPotentialConsumers.containsKey(partition)) {
                        // if this topic partition of this consumer no longer exists remove it from currentAssignment of the consumer
                        partitionIter.remove();
                        currentPartitionConsumer.remove(partition);
                    } else if (!subscriptions.get(entry.getKey()).topics().contains(partition.topic())) {
                        // if this partition cannot remain assigned to its current consumer because the consumer
                        // is no longer subscribed to its topic remove it from currentAssignment of the consumer
                        partitionIter.remove();
                    } else
                        // otherwise, remove the topic partition from those that need to be assigned only if
                        // its current consumer is still subscribed to its topic (because it is already assigned
                        // and we would want to preserve that assignment as much as possible)
                        unassignedPartitions.remove(partition);
                }
            }
        }
        // at this point we have preserved all valid topic partition to consumer assignments and removed
        // all invalid topic partitions and invalid consumers. Now we need to assign unassignedPartitions
        // to consumers so that the topic partition assignments are as balanced as possible.

        // an ascending sorted set of consumers based on how many topic partitions are already assigned to them
        TreeSet<String> sortedCurrentSubscriptions = new TreeSet<>(new SubscriptionComparator(currentAssignment));
        sortedCurrentSubscriptions.addAll(currentAssignment.keySet());

        balance(currentAssignment, sortedPartitions, unassignedPartitions, sortedCurrentSubscriptions,
                consumer2AllPotentialPartitions, partition2AllPotentialConsumers, currentPartitionConsumer);
        return currentAssignment;
    }

    private void prepopulateCurrentAssignments(Map<String, Subscription> subscriptions,
                                               Map<String, List<TopicPartition>> currentAssignment) {
        for (Map.Entry<String, Subscription> subscriptionEntry : subscriptions.entrySet()) {
            ByteBuffer userData = subscriptionEntry.getValue().userData();
            if (userData != null && userData.hasRemaining())
                currentAssignment.put(subscriptionEntry.getKey(), deserializeTopicPartitionAssignment(userData));
        }
    }

    @Override
    public void onAssignment(Assignment assignment) {
        memberAssignment = assignment.partitions();
    }

    @Override
    public Subscription subscription(Set<String> topics) {
        if (memberAssignment == null)
            return new Subscription(new ArrayList<>(topics));

        return new Subscription(new ArrayList<>(topics), serializeTopicPartitionAssignment(memberAssignment));
    }

    @Override
    public String name() {
        return "sticky";
    }

    /**
     * determine if the current assignment is a balanced one
     *
     * @param currentAssignment: the assignment whose balance needs to be checked
     * @param sortedCurrentSubscriptions: an ascending sorted set of consumers based on how many topic partitions are already assigned to them
     * @param allSubscriptions: a mapping of all consumers to all potential topic partitions that can be assigned to them
     * @return true if the given assignment is balanced; false otherwise
     */
    private boolean isBalanced(Map<String, List<TopicPartition>> currentAssignment,
                               TreeSet<String> sortedCurrentSubscriptions,
                               Map<String, List<TopicPartition>> allSubscriptions) {
        int min = currentAssignment.get(sortedCurrentSubscriptions.first()).size();
        int max = currentAssignment.get(sortedCurrentSubscriptions.last()).size();
        if (min >= max - 1)
            // if minimum and maximum numbers of partitions assigned to consumers differ by at most one return true
            return true;

        // create a mapping from partitions to the consumer assigned to them
        final Map<TopicPartition, String> allPartitions = new HashMap<>();
        Set<Entry<String, List<TopicPartition>>> assignments = currentAssignment.entrySet();
        for (Map.Entry<String, List<TopicPartition>> entry: assignments) {
            List<TopicPartition> topicPartitions = entry.getValue();
            for (TopicPartition topicPartition: topicPartitions) {
                if (allPartitions.containsKey(topicPartition))
                    log.error(topicPartition + " is assigned to more than one consumer.");
                allPartitions.put(topicPartition, entry.getKey());
            }
        }

        // for each consumer that does not have all the topic partitions it can get make sure none of the topic partitions it
        // could but did not get cannot be moved to it (because that would break the balance)
        for (String consumer: sortedCurrentSubscriptions) {
            List<TopicPartition> consumerPartitions = currentAssignment.get(consumer);
            int consumerPartitionCount = consumerPartitions.size();

            // skip if this consumer already has all the topic partitions it can get
            if (consumerPartitionCount == allSubscriptions.get(consumer).size())
                continue;

            // otherwise make sure it cannot get any more
            List<TopicPartition> potentialTopicPartitions = allSubscriptions.get(consumer);
            for (TopicPartition topicPartition: potentialTopicPartitions) {
                if (!currentAssignment.get(consumer).contains(topicPartition)) {
                    String otherConsumer = allPartitions.get(topicPartition);
                    int otherConsumerPartitionCount = currentAssignment.get(otherConsumer).size();
                    if (consumerPartitionCount < otherConsumerPartitionCount) {
                        log.debug(topicPartition + " can be moved from consumer " + otherConsumer + " to consumer " + consumer + " for a more balanced assignment.");
                        return false;
                    }
                }
            }
        }
        return true;
    }

    /**
     * @return the balance score of the given assignment, as the sum of assigned partitions size difference of all consumer pairs.
     * A perfectly balanced assignment (with all consumers getting the same number of partitions) has a balance score of 0.
     * Lower balance score indicates a more balanced assignment.
     */
    private int getBalanceScore(Map<String, List<TopicPartition>> assignment) {
        int score = 0;

        Map<String, Integer> consumer2AssignmentSize = new HashMap<>();
        for (Entry<String, List<TopicPartition>> entry: assignment.entrySet())
            consumer2AssignmentSize.put(entry.getKey(), entry.getValue().size());

        Iterator<Entry<String, Integer>> it = consumer2AssignmentSize.entrySet().iterator();
        while (it.hasNext()) {
            Entry<String, Integer> entry = it.next();
            int consumerAssignmentSize = entry.getValue();
            it.remove();
            for (Entry<String, Integer> otherEntry: consumer2AssignmentSize.entrySet())
                score += Math.abs(consumerAssignmentSize - otherEntry.getValue());
        }

        return score;
    }

    /**
     * Sort valid partitions so they are processed in the potential reassignment phase in the proper order
     * that causes minimal partition movement among consumers (hence honoring maximal stickiness)
     *
     * @param currentAssignment the calculated assignment so far
     * @param isFreshAssignment whether this is a new assignment, or a reassignment of an existing one
     * @param partition2AllPotentialConsumers a mapping of partitions to their potential consumers
     * @param consumer2AllPotentialPartitions a mapping of consumers to potential partitions they can consumer from
     * @return sorted list of valid partitions
     */
    private List<TopicPartition> sortPartitions(Map<String, List<TopicPartition>> currentAssignment,
                                                boolean isFreshAssignment,
                                                Map<TopicPartition, List<String>> partition2AllPotentialConsumers,
                                                Map<String, List<TopicPartition>> consumer2AllPotentialPartitions) {
        List<TopicPartition> sortedPartitions = new ArrayList<>();

        if (!isFreshAssignment && areSubscriptionsIdentical(partition2AllPotentialConsumers, consumer2AllPotentialPartitions)) {
            // if this is a reassignment and the subscriptions are identical (all consumers can consumer from all topics)
            // then we just need to simply list partitions in a round robin fashion (from consumers with
            // most assigned partitions to those with least)
            Map<String, List<TopicPartition>> assignments = deepCopy(currentAssignment);
            for (Entry<String, List<TopicPartition>> entry: assignments.entrySet()) {
                List<TopicPartition> toRemove = new ArrayList<>();
                for (TopicPartition partition: entry.getValue())
                    if (!partition2AllPotentialConsumers.keySet().contains(partition))
                        toRemove.add(partition);
                for (TopicPartition partition: toRemove)
                    entry.getValue().remove(partition);
            }
            TreeSet<String> sortedConsumers = new TreeSet<>(new SubscriptionComparator(assignments));
            sortedConsumers.addAll(assignments.keySet());

            while (!sortedConsumers.isEmpty()) {
                String consumer = sortedConsumers.pollLast();
                List<TopicPartition> remainingPartitions = assignments.get(consumer);
                if (!remainingPartitions.isEmpty()) {
                    sortedPartitions.add(remainingPartitions.remove(0));
                    sortedConsumers.add(consumer);
                }
            }

            for (TopicPartition partition: partition2AllPotentialConsumers.keySet()) {
                if (!sortedPartitions.contains(partition))
                    sortedPartitions.add(partition);
            }

        } else {
            // an ascending sorted set of topic partitions based on how many consumers can potentially use them
            TreeSet<TopicPartition> sortedAllPartitions = new TreeSet<>(new PartitionComparator(partition2AllPotentialConsumers));
            sortedAllPartitions.addAll(partition2AllPotentialConsumers.keySet());

            while (!sortedAllPartitions.isEmpty())
                sortedPartitions.add(sortedAllPartitions.pollFirst());
        }

        return sortedPartitions;
    }

    /**
     * @param partition2AllPotentialConsumers a mapping of partitions to their potential consumers
     * @param consumer2AllPotentialPartitions a mapping of consumers to potential partitions they can consumer from
     * @return true if potential consumers of partitions are the same, and potential partitions consumers can
     * consumer from are the same too
     */
    private boolean areSubscriptionsIdentical(Map<TopicPartition, List<String>> partition2AllPotentialConsumers,
                                              Map<String, List<TopicPartition>> consumer2AllPotentialPartitions) {
        if (!hasIdenticalListElements(partition2AllPotentialConsumers.values()))
            return false;

        if (!hasIdenticalListElements(consumer2AllPotentialPartitions.values()))
            return false;

        return true;
    }

    /**
     * @return the consumer to which the given partition is assigned. The assignment should improve the overall balance
     * of the partition assignments to consumers.
     */
    private String assignPartition(TopicPartition partition,
                                   TreeSet<String> sortedCurrentSubscriptions,
                                   Map<String, List<TopicPartition>> currentAssignment,
                                   Map<String, List<TopicPartition>> consumer2AllPotentialPartitions,
                                   Map<TopicPartition, String> currentPartitionConsumer) {
        for (String consumer: sortedCurrentSubscriptions) {
            if (consumer2AllPotentialPartitions.get(consumer).contains(partition)) {
                sortedCurrentSubscriptions.remove(consumer);
                currentAssignment.get(consumer).add(partition);
                currentPartitionConsumer.put(partition, consumer);
                sortedCurrentSubscriptions.add(consumer);
                return consumer;
            }
        }
        return null;
    }

    private boolean canParticipateInReassignment(TopicPartition partition,
                                                 Map<TopicPartition, List<String>> partition2AllPotentialConsumers) {
        // if a partition has two or more potential consumers it is subject to reassignment.
        return partition2AllPotentialConsumers.get(partition).size() >= 2;
    }

    private boolean canParticipateInReassignment(String consumer,
                                                 Map<String, List<TopicPartition>> currentAssignment,
                                                 Map<String, List<TopicPartition>> consumer2AllPotentialPartitions,
                                                 Map<TopicPartition, List<String>> partition2AllPotentialConsumers) {
        List<TopicPartition> currentPartitions = currentAssignment.get(consumer);
        int currentAssignmentSize = currentPartitions.size();
        int maxAssignmentSize = consumer2AllPotentialPartitions.get(consumer).size();
        if (currentAssignmentSize > maxAssignmentSize)
            log.error("The consumer " + consumer + " is assigned more partitions than the maximum possible.");

        if (currentAssignmentSize < maxAssignmentSize)
            // if a consumer is not assigned all its potential partitions it is subject to reassignment
            return true;

        for (TopicPartition partition: currentPartitions)
            // if any of the partitions assigned to a consumer is subject to reassignment the consumer itself
            // is subject to reassignment
            if (canParticipateInReassignment(partition, partition2AllPotentialConsumers))
                return true;

        return false;
    }

    /**
     * Balance the current assignment using the data structures created in the assign(...) method above.
     */
    private void balance(Map<String, List<TopicPartition>> currentAssignment,
                         List<TopicPartition> sortedPartitions,
                         List<TopicPartition> unassignedPartitions,
                         TreeSet<String> sortedCurrentSubscriptions,
                         Map<String, List<TopicPartition>> consumer2AllPotentialPartitions,
                         Map<TopicPartition, List<String>> partition2AllPotentialConsumers,
                         Map<TopicPartition, String> currentPartitionConsumer) {
        boolean initializing = currentAssignment.get(sortedCurrentSubscriptions.last()).isEmpty();
        boolean reassignmentPerformed = false;

        // assign all unassigned partitions
        for (TopicPartition partition: unassignedPartitions) {
            // skip if there is no potential consumer for the partition
            if (partition2AllPotentialConsumers.get(partition).isEmpty())
                continue;

            assignPartition(partition, sortedCurrentSubscriptions, currentAssignment,
                            consumer2AllPotentialPartitions, currentPartitionConsumer);
        }

        // narrow down the reassignment scope to only those partitions that can actually be reassigned
        Set<TopicPartition> fixedPartitions = new HashSet<>();
        for (TopicPartition partition: partition2AllPotentialConsumers.keySet())
            if (!canParticipateInReassignment(partition, partition2AllPotentialConsumers))
                fixedPartitions.add(partition);
        sortedPartitions.removeAll(fixedPartitions);

        // narrow down the reassignment scope to only those consumers that are subject to reassignment
        Map<String, List<TopicPartition>> fixedAssignments = new HashMap<>();
        for (String consumer: consumer2AllPotentialPartitions.keySet())
            if (!canParticipateInReassignment(consumer, currentAssignment,
                                              consumer2AllPotentialPartitions, partition2AllPotentialConsumers)) {
                sortedCurrentSubscriptions.remove(consumer);
                fixedAssignments.put(consumer, currentAssignment.remove(consumer));
            }

        // create a deep copy of the current assignment so we can revert to it if we do not get a more balanced assignment later
        Map<String, List<TopicPartition>> preBalanceAssignment = deepCopy(currentAssignment);
        Map<TopicPartition, String> preBalancePartitionConsumers = new HashMap<>(currentPartitionConsumer);

        reassignmentPerformed = performReassignments(sortedPartitions, currentAssignment, sortedCurrentSubscriptions,
                consumer2AllPotentialPartitions, partition2AllPotentialConsumers, currentPartitionConsumer);

        // if we are not preserving existing assignments and we have made changes to the current assignment
        // make sure we are getting a more balanced assignment; otherwise, revert to previous assignment
        if (!initializing && reassignmentPerformed && getBalanceScore(currentAssignment) >= getBalanceScore(preBalanceAssignment)) {
            deepCopy(preBalanceAssignment, currentAssignment);
            currentPartitionConsumer.clear();
            currentPartitionConsumer.putAll(preBalancePartitionConsumers);
        }

        // add the fixed assignments (those that could not change) back
        for (Entry<String, List<TopicPartition>> entry: fixedAssignments.entrySet()) {
            String consumer = entry.getKey();
            currentAssignment.put(consumer, entry.getValue());
            sortedCurrentSubscriptions.add(consumer);
        }

        fixedAssignments.clear();
    }

    private boolean performReassignments(List<TopicPartition> reassignablePartitions,
                                         Map<String, List<TopicPartition>> currentAssignment,
                                         TreeSet<String> sortedCurrentSubscriptions,
                                         Map<String, List<TopicPartition>> consumer2AllPotentialPartitions,
                                         Map<TopicPartition, List<String>> partition2AllPotentialConsumers,
                                         Map<TopicPartition, String> currentPartitionConsumer) {
        boolean reassignmentPerformed = false;
        boolean modified;

        // repeat reassignment until no partition can be moved to improve the balance
        do {
            modified = false;
            // reassign all reassignable partitions (starting from the partition with least potential consumers and if needed)
            // until the full list is processed or a balance is achieved
            Iterator<TopicPartition> partitionIterator = reassignablePartitions.iterator();
            while (partitionIterator.hasNext() && !isBalanced(currentAssignment, sortedCurrentSubscriptions, consumer2AllPotentialPartitions)) {
                TopicPartition partition = partitionIterator.next();

                // the partition must have at least two consumers
                if (partition2AllPotentialConsumers.get(partition).size() <= 1)
                    log.error("Expected more than one potential consumer for partition '" + partition + "'");

                // the partition must have a current consumer
                String consumer = currentPartitionConsumer.get(partition);
                if (consumer == null)
                    log.error("Expected partition '" + partition + "' to be assigned to a consumer");

                // check if a better-suited consumer exist for the partition; if so, reassign it
                for (String otherConsumer: partition2AllPotentialConsumers.get(partition)) {
                    if (currentAssignment.get(consumer).size() > currentAssignment.get(otherConsumer).size() + 1) {
                        reassignPartition(partition, currentAssignment, sortedCurrentSubscriptions, currentPartitionConsumer, consumer2AllPotentialPartitions);
                        reassignmentPerformed = true;
                        modified = true;
                        break;
                    }
                }
            }
        } while (modified);

        return reassignmentPerformed;
    }

    private void reassignPartition(TopicPartition partition,
                                   Map<String, List<TopicPartition>> currentAssignment,
                                   TreeSet<String> sortedCurrentSubscriptions,
                                   Map<TopicPartition, String> currentPartitionConsumer,
                                   Map<String, List<TopicPartition>> consumer2AllPotentialPartitions) {
        String consumer = currentPartitionConsumer.get(partition);

        // find the new consumer
        String newConsumer = null;
        for (String anotherConsumer: sortedCurrentSubscriptions) {
            if (consumer2AllPotentialPartitions.get(anotherConsumer).contains(partition)) {
                newConsumer = anotherConsumer;
                break;
            }
        }

        assert newConsumer != null;

        // find the correct partition movement considering the stickiness requirement
        TopicPartition partitionToBeMoved = partitionMovements.getTheActualPartitionToBeMoved(partition, consumer, newConsumer);
        processPartitionMovement(partitionToBeMoved, newConsumer, currentAssignment, sortedCurrentSubscriptions, currentPartitionConsumer);

        return;
    }

    private void processPartitionMovement(TopicPartition partition,
                                          String newConsumer,
                                          Map<String, List<TopicPartition>> currentAssignment,
                                          TreeSet<String> sortedCurrentSubscriptions,
                                          Map<TopicPartition, String> currentPartitionConsumer) {
        String oldConsumer = currentPartitionConsumer.get(partition);

        sortedCurrentSubscriptions.remove(oldConsumer);
        sortedCurrentSubscriptions.remove(newConsumer);

        partitionMovements.movePartition(partition, oldConsumer, newConsumer);

        currentAssignment.get(oldConsumer).remove(partition);
        currentAssignment.get(newConsumer).add(partition);
        currentPartitionConsumer.put(partition, newConsumer);
        sortedCurrentSubscriptions.add(newConsumer);
        sortedCurrentSubscriptions.add(oldConsumer);
    }

    boolean isSticky() {
        return partitionMovements.isSticky();
    }

    static ByteBuffer serializeTopicPartitionAssignment(List<TopicPartition> partitions) {
        Struct struct = new Struct(STICKY_ASSIGNOR_USER_DATA);
        List<Struct> topicAssignments = new ArrayList<>();
        for (Map.Entry<String, List<Integer>> topicEntry : CollectionUtils.groupPartitionsByTopic(partitions).entrySet()) {
            Struct topicAssignment = new Struct(TOPIC_ASSIGNMENT);
            topicAssignment.set(TOPIC_KEY_NAME, topicEntry.getKey());
            topicAssignment.set(PARTITIONS_KEY_NAME, topicEntry.getValue().toArray());
            topicAssignments.add(topicAssignment);
        }
        struct.set(TOPIC_PARTITIONS_KEY_NAME, topicAssignments.toArray());
        ByteBuffer buffer = ByteBuffer.allocate(STICKY_ASSIGNOR_USER_DATA.sizeOf(struct));
        STICKY_ASSIGNOR_USER_DATA.write(buffer, struct);
        buffer.flip();
        return buffer;
    }

    private static List<TopicPartition> deserializeTopicPartitionAssignment(ByteBuffer buffer) {
        Struct struct = STICKY_ASSIGNOR_USER_DATA.read(buffer);
        List<TopicPartition> partitions = new ArrayList<>();
        for (Object structObj : struct.getArray(TOPIC_PARTITIONS_KEY_NAME)) {
            Struct assignment = (Struct) structObj;
            String topic = assignment.getString(TOPIC_KEY_NAME);
            for (Object partitionObj : assignment.getArray(PARTITIONS_KEY_NAME)) {
                Integer partition = (Integer) partitionObj;
                partitions.add(new TopicPartition(topic, partition));
            }
        }
        return partitions;
    }

    /**
     * @param col a collection of elements of type list
     * @return true if all lists in the collection have the same members; false otherwise
     */
    private <T> boolean hasIdenticalListElements(Collection<List<T>> col) {
        Iterator<List<T>> it = col.iterator();
        List<T> cur = it.next();
        while (it.hasNext()) {
            List<T> next = it.next();
            if (!(cur.containsAll(next) && next.containsAll(cur)))
                return false;
            cur = next;
        }
        return true;
    }

    private void deepCopy(Map<String, List<TopicPartition>> source, Map<String, List<TopicPartition>> dest) {
        dest.clear();
        for (Entry<String, List<TopicPartition>> entry: source.entrySet())
            dest.put(entry.getKey(), new ArrayList<>(entry.getValue()));
    }

    private Map<String, List<TopicPartition>> deepCopy(Map<String, List<TopicPartition>> assignment) {
        Map<String, List<TopicPartition>> copy = new HashMap<>();
        deepCopy(assignment, copy);
        return copy;
    }

    private static class PartitionComparator implements Comparator<TopicPartition>, Serializable {
        private static final long serialVersionUID = 1L;
        private Map<TopicPartition, List<String>> map;

        PartitionComparator(Map<TopicPartition, List<String>> map) {
            this.map = map;
        }

        @Override
        public int compare(TopicPartition o1, TopicPartition o2) {
            int ret = map.get(o1).size() - map.get(o2).size();
            if (ret == 0) {
                ret = o1.topic().compareTo(o2.topic());
                if (ret == 0)
                    ret = o1.partition() - o2.partition();
            }
            return ret;
        }
    }

    private static class SubscriptionComparator implements Comparator<String>, Serializable {
        private static final long serialVersionUID = 1L;
        private Map<String, List<TopicPartition>> map;

        SubscriptionComparator(Map<String, List<TopicPartition>> map) {
            this.map = map;
        }

        @Override
        public int compare(String o1, String o2) {
            int ret = map.get(o1).size() - map.get(o2).size();
            if (ret == 0)
                ret = o1.compareTo(o2);
            return ret;
        }
    }

    /**
     * This class maintains some data structures to simplify lookup of partition movements among consumers. At each point of
     * time during a partition rebalance it keeps track of partition movements corresponding to each topic, and also possible
     * movement (in form a ConsumerPair object) for each partition.
     */
    private static class PartitionMovements {
        private Map<String, Map<ConsumerPair, Set<TopicPartition>>> partitionMovementsByTopic = new HashMap<>();
        private Map<TopicPartition, ConsumerPair> partitionMovements = new HashMap<>();

        private ConsumerPair removeMovementRecordOfPartition(TopicPartition partition) {
            ConsumerPair pair = partitionMovements.remove(partition);

            String topic = partition.topic();
            Map<ConsumerPair, Set<TopicPartition>> partitionMovementsForThisTopic = partitionMovementsByTopic.get(topic);
            partitionMovementsForThisTopic.get(pair).remove(partition);
            if (partitionMovementsForThisTopic.get(pair).isEmpty())
                partitionMovementsForThisTopic.remove(pair);
            if (partitionMovementsByTopic.get(topic).isEmpty())
                partitionMovementsByTopic.remove(topic);

            return pair;
        }

        private void addPartitionMovementRecord(TopicPartition partition, ConsumerPair pair) {
            partitionMovements.put(partition, pair);

            String topic = partition.topic();
            if (!partitionMovementsByTopic.containsKey(topic))
                partitionMovementsByTopic.put(topic, new HashMap<ConsumerPair, Set<TopicPartition>>());

            Map<ConsumerPair, Set<TopicPartition>> partitionMovementsForThisTopic = partitionMovementsByTopic.get(topic);
            if (!partitionMovementsForThisTopic.containsKey(pair))
                partitionMovementsForThisTopic.put(pair, new HashSet<TopicPartition>());

            partitionMovementsForThisTopic.get(pair).add(partition);
        }

        private void movePartition(TopicPartition partition, String oldConsumer, String newConsumer) {
            ConsumerPair pair = new ConsumerPair(oldConsumer, newConsumer);

            if (partitionMovements.containsKey(partition)) {
                // this partition has previously moved
                ConsumerPair existingPair = removeMovementRecordOfPartition(partition);
                assert existingPair.dstMemberId.equals(oldConsumer);
                if (!existingPair.srcMemberId.equals(newConsumer)) {
                    // the partition is not moving back to its previous consumer
                    // return new ConsumerPair2(existingPair.src, newConsumer);
                    addPartitionMovementRecord(partition, new ConsumerPair(existingPair.srcMemberId, newConsumer));
                }
            } else
                addPartitionMovementRecord(partition, pair);
        }

        private TopicPartition getTheActualPartitionToBeMoved(TopicPartition partition, String oldConsumer, String newConsumer) {
            String topic = partition.topic();

            if (!partitionMovementsByTopic.containsKey(topic))
                return partition;

            if (partitionMovements.containsKey(partition)) {
                // this partition has previously moved
                assert oldConsumer.equals(partitionMovements.get(partition).dstMemberId);
                oldConsumer = partitionMovements.get(partition).srcMemberId;
            }

            Map<ConsumerPair, Set<TopicPartition>> partitionMovementsForThisTopic = partitionMovementsByTopic.get(topic);
            ConsumerPair reversePair = new ConsumerPair(newConsumer, oldConsumer);
            if (!partitionMovementsForThisTopic.containsKey(reversePair))
                return partition;

            return partitionMovementsForThisTopic.get(reversePair).iterator().next();
        }

        private boolean isLinked(String src, String dst, Set<ConsumerPair> pairs, List<String> currentPath) {
            if (src.equals(dst))
                return false;

            if (pairs.isEmpty())
                return false;

            if (new ConsumerPair(src, dst).in(pairs)) {
                currentPath.add(src);
                currentPath.add(dst);
                return true;
            }

            for (ConsumerPair pair: pairs)
                if (pair.srcMemberId.equals(src)) {
                    Set<ConsumerPair> reducedSet = new HashSet<>(pairs);
                    reducedSet.remove(pair);
                    currentPath.add(pair.srcMemberId);
                    return isLinked(pair.dstMemberId, dst, reducedSet, currentPath);
                }

            return false;
        }

        private boolean in(List<String> cycle, Set<List<String>> cycles) {
            List<String> superCycle = new ArrayList<>(cycle);
            superCycle.remove(superCycle.size() - 1);
            superCycle.addAll(cycle);
            for (List<String> foundCycle: cycles) {
                if (foundCycle.size() == cycle.size() && Collections.indexOfSubList(superCycle, foundCycle) != -1)
                    return true;
            }
            return false;
        }

        private boolean hasCycles(Set<ConsumerPair> pairs) {
            Set<List<String>> cycles = new HashSet<>();
            for (ConsumerPair pair: pairs) {
                Set<ConsumerPair> reducedPairs = new HashSet<>(pairs);
                reducedPairs.remove(pair);
                List<String> path = new ArrayList<>(Collections.singleton(pair.srcMemberId));
                if (isLinked(pair.dstMemberId, pair.srcMemberId, reducedPairs, path) && !in(path, cycles)) {
                    cycles.add(new ArrayList<>(path));
                    log.error("A cycle of length " + (path.size() - 1) + " was found: " + path.toString());
                }
            }

            // for now we want to make sure there is no partition movements of the same topic between a pair of consumers.
            // the odds of finding a cycle among more than two consumers seem to be very low (according to various randomized
            // tests with the given sticky algorithm) that it should not worth the added complexity of handling those cases.
            for (List<String> cycle: cycles)
                if (cycle.size() == 3) // indicates a cycle of length 2
                    return true;
            return false;
        }

        private boolean isSticky() {
            for (Map.Entry<String, Map<ConsumerPair, Set<TopicPartition>>> topicMovements: this.partitionMovementsByTopic.entrySet()) {
                Set<ConsumerPair> topicMovementPairs = topicMovements.getValue().keySet();
                if (hasCycles(topicMovementPairs)) {
                    log.error("Stickiness is violated for topic " + topicMovements.getKey()
                            + "\nPartition movements for this topic occurred among the following consumer pairs:"
                            + "\n" + topicMovements.getValue().toString());
                    return false;
                }
            }

            return true;
        }
    }

    /**
     * ConsumerPair represents a pair of Kafka consumer ids involved in a partition reassignment. Each
     * ConsumerPair object, which contains a source (src) and a destination (dst)
     * element, normally corresponds to a particular partition or topic, and indicates that the particular partition or some
     * partition of the particular topic was moved from the source consumer to the destination consumer during the rebalance.
     * This class is used, through the PartitionMovements class, by the sticky assignor and helps in determining
     * whether a partition reassignment results in cycles among the generated graph of consumer pairs.
     */
    private static class ConsumerPair {
        private final String srcMemberId;
        private final String dstMemberId;

        ConsumerPair(String srcMemberId, String dstMemberId) {
            this.srcMemberId = srcMemberId;
            this.dstMemberId = dstMemberId;
        }

        public String toString() {
            return this.srcMemberId + "->" + this.dstMemberId;
        }

        @Override
        public int hashCode() {
            final int prime = 31;
            int result = 1;
            result = prime * result + ((this.srcMemberId == null) ? 0 : this.srcMemberId.hashCode());
            result = prime * result + ((this.dstMemberId == null) ? 0 : this.dstMemberId.hashCode());
            return result;
        }

        @Override
        public boolean equals(Object obj) {
            if (obj == null)
                return false;

            if (!getClass().isInstance(obj))
                return false;

            ConsumerPair otherPair = (ConsumerPair) obj;
            return this.srcMemberId.equals(otherPair.srcMemberId) && this.dstMemberId.equals(otherPair.dstMemberId);
        }

        private boolean in(Set<ConsumerPair> pairs) {
            for (ConsumerPair pair: pairs)
                if (this.equals(pair))
                    return true;
            return false;
        }
    }
}


text:
<?php

namespace Magento\Widget\Controller\Adminhtml\Widget\Instance;

class NewAction extends \Magento\Widget\Controller\Adminhtml\Widget\Instance
{
    /**
     * New widget instance action (forward to edit action)
     *
     * @return void
     */
    public function execute()
    {
        $this->_forward('edit');
    }
}


text:
using System;
using Android.App;
using Android.Widget;
using Android.OS;
using LBXamarinSDK;
using LBXamarinSDK.LBRepo;
using System.Timers;
using System.Collections.Generic;
using System.Linq;
using ExampleApp;

namespace Agents
{
	[Activity (Label = "Loopback SDK Usage Example", MainLauncher = true, Icon = "@drawable/icon")]
	public class MainActivity : Activity
	{
		bool connected = false;
		System.Timers.Timer connectedTimer;
		System.Timers.Timer dataReloadTimer;
		IList<Mission> missions = new List<Mission>();

		Button loginButton;
		Button logoutButton;
		Button missionCreateButton;
		ListView listView;
		TextView missionPriorityText;
		TextView geoLocationText;
		EditText passwordText;
		EditText missionDescriptionText;
		EditText priorityText;
		EditText latitudeText;
		EditText longitudeText;
		TextView creationStatus;
		TextView connectedStatus;
		EditText loginText;
		TextView loginStatus;

		/**
		 * Initializes all references to a particular view component
		 */
		private void initUiReferences()
		{
			loginButton = FindViewById<Button> (Resource.Id.button1);
			logoutButton = FindViewById<Button> (Resource.Id.button2);
			missionCreateButton = FindViewById<Button> (Resource.Id.button3);
			listView = FindViewById<ListView> (Resource.Id.listView1);
			loginStatus = FindViewById<TextView> (Resource.Id.textView3);
			missionPriorityText = FindViewById<TextView> (Resource.Id.textView7);
			geoLocationText = FindViewById<TextView> (Resource.Id.textView8);
			connectedStatus = FindViewById<TextView> (Resource.Id.textView5);
			creationStatus = FindViewById<TextView> (Resource.Id.textView13);
			passwordText = FindViewById<EditText> (Resource.Id.editText1);
			loginText = FindViewById<EditText> (Resource.Id.editText2);
			missionDescriptionText = FindViewById<EditText> (Resource.Id.editText3);
			priorityText = FindViewById<EditText> (Resource.Id.editText4);
			latitudeText = FindViewById<EditText> (Resource.Id.editText5);
			longitudeText = FindViewById<EditText> (Resource.Id.editText6);
		}

		protected override void OnCreate (Bundle bundle)
		{
			base.OnCreate (bundle);
			SetContentView (Resource.Layout.Main);
			initUiReferences ();

			Gateway.SetDebugMode (true);
			Gateway.SetServerBaseURL (new Uri("http://10.0.0.30:3000/api/"));

			loginButton.Click += doLogin;
			logoutButton.Click += doLogout;
			missionCreateButton.Click += createMission;

			var listAdapter = new ArrayAdapter<string>(this, Android.Resource.Layout.SimpleListItem1, new List<string>());
			listView.Adapter = listAdapter;
			listView.ItemClick += delegate(object sender, AdapterView.ItemClickEventArgs e)
			{
				displayMissionDetails(e.Id);
			};

			updateMissions (null, null);
			StartMissionUpdateTimer ();
			StartConnectedIntervalChecks ();
		}

		public void displayMissionDetails(long id)
		{
			Mission clickedOnMission = missions.ElementAt ((int)id);

			const int priorityRedThreshold = 3;

			if (clickedOnMission.priority < priorityRedThreshold)
			{
				missionPriorityText.SetTextColor (Android.Graphics.Color.Red);
			}
			else
			{
				missionPriorityText.SetTextColor (Android.Graphics.Color.White);
			}

			missionPriorityText.Text = "Mission: " + clickedOnMission.description + " Priority " + clickedOnMission.priority.ToString();

			if (clickedOnMission.location != null)
			{
				geoLocationText.Text = "The mission " + clickedOnMission.description + " will take place at coordinates ("
					+ clickedOnMission.location.Longitude + ", " + clickedOnMission.location.Latitude + ")";
			}
		}

		public async void createMission(Object sender, EventArgs e)
		{
			Mission createdMission = new Mission ()
			{
				description = missionDescriptionText.Text
			};

			if (!string.IsNullOrEmpty(priorityText.Text))
			{
				createdMission.priority = int.Parse (priorityText.Text);
			}

			if (!string.IsNullOrEmpty(longitudeText.Text) && !string.IsNullOrEmpty(latitudeText.Text))
			{
				createdMission.location = new GeoPoint ()
				{
					Latitude = int.Parse (latitudeText.Text),
					Longitude = int.Parse (longitudeText.Text)
				};
			}

			try
			{
				createdMission = await Missions.Create (createdMission);

				creationStatus.SetTextColor(Android.Graphics.Color.Green);
				creationStatus.Text = "Mission created successfully.";
				missionDescriptionText.Text = "";
				priorityText.Text = "";
				latitudeText.Text = "";
				longitudeText.Text = "";
				updateMissions (null, null);

			}
			catch(RestException exception)
			{
				creationStatus.SetTextColor(Android.Graphics.Color.Red);
				creationStatus.Text = "Creation failed.";

				if (exception.StatusCode == 401)
				{
					Toast.MakeText (this, "Creation failed: You are not authorized to create, please log in.", ToastLength.Short).Show ();
				}
				else if (exception.StatusCode == 422)
				{
					Toast.MakeText (this, "Creation failed: Please make sure that you've filled all the required mission details.", ToastLength.Short).Show ();
				}
				else
				{
					Toast.MakeText (this, "Creation failed" + exception.StatusCode.ToString(), ToastLength.Short).Show ();
				}
			}
		}

		public async void updateMissions(object source, ElapsedEventArgs e)
		{
			RunOnUiThread (async delegate
				{
					try
					{
						missions = await Missions.Find ();
						ArrayAdapter<string> listAdapter = listView.Adapter as ArrayAdapter<string>;
						listAdapter.Clear();
						listAdapter.AddAll(missions.Select(x => x.description).ToList());
					}
					catch(Exception)
					{
						// All loopback repositories throw RestException on failure.
						// This is a placeholder for logic on this failure of missions.find
					}
			});
		}

		public void StartMissionUpdateTimer()
		{
			dataReloadTimer = new System.Timers.Timer(3000);
			dataReloadTimer.Elapsed += new ElapsedEventHandler(updateMissions);
			dataReloadTimer.Interval = 3000;
			dataReloadTimer.Enabled = true;
			dataReloadTimer.Start();
		}

		public void StartConnectedIntervalChecks()
		{
			connectedTimer = new System.Timers.Timer(3000);
			connectedTimer.Elapsed += new ElapsedEventHandler(updateConnectedStatus);
			connectedTimer.Interval = 3000;
			connectedTimer.Enabled = true;
			connectedTimer.Start();
		}

		public async void updateConnectedStatus(object source, ElapsedEventArgs e)
		{

			connected =  await Gateway.isConnected (500);
			RunOnUiThread(delegate {
				if(connected)
				{
					connectedStatus.Text = "Status: Connected to Server";
					connectedStatus.SetTextColor(Android.Graphics.Color.Green);
					UnlockEntireUI ();
				}
				else {
					connectedStatus.Text = "Status: Disconnected from Server";
					connectedStatus.SetTextColor(Android.Graphics.Color.Red);
					LockEntireUI ();
				}
			});
		}

		public void LockEntireUI()
		{
			missionCreateButton.Enabled = false;
			logoutButton.Enabled = false;
			passwordText.Enabled = false;
			loginText.Enabled = false;
			loginButton.Enabled = false;
			loginStatus.SetTextColor(Android.Graphics.Color.Red);
			loginStatus.Text = "Logged out.";
		}

		public void UnlockEntireUI()
		{
			missionCreateButton.Enabled = true;
			if(loginButton.Enabled == true || logoutButton.Enabled == false)
			{
				logoutButton.Enabled = true;
				passwordText.Enabled = true;
				loginText.Enabled = true;
				loginButton.Enabled = true;
			}

		}

		public void lockLoginUI()
		{
			loginStatus.SetTextColor(Android.Graphics.Color.Green);
			loginStatus.Text = "Logged in!";
			passwordText.Enabled = false;
			loginText.Enabled = false;
			loginButton.Enabled = false;
		}

		public void UnlockLoginUI()
		{
			loginStatus.SetTextColor(Android.Graphics.Color.Red);
			loginStatus.Text = "Logged out.";
			loginText.Text = "";
			passwordText.Text = "";
			passwordText.Enabled = true;
			loginText.Enabled = true;
			loginButton.Enabled = true;
		}

		public async void doLogin(Object sender, EventArgs e)
		{
			if (string.IsNullOrEmpty(passwordText.Text) || string.IsNullOrEmpty(loginText.Text))
			{
				Toast.MakeText (this, "Can't login with empty credentials.", ToastLength.Short).Show ();
				return;
			}

			try
			{
				Operator loginCredentials = new Operator ()
				{
					password = passwordText.Text,
					email = loginText.Text
				};
				AccessToken accessToken = await Operators.login (loginCredentials);

				lockLoginUI();
				Gateway.SetAccessToken(accessToken);
			}
			catch(Exception)
			{
				Toast.MakeText (this, "Login failed.", ToastLength.Short).Show ();
				UnlockLoginUI ();
			}
		}

		public async void doLogout(Object sender, EventArgs e)
		{
			try
			{
				await Operators.logout ();
			}
			catch(RestException)
			{
				Toast.MakeText (this, "You are not logged in, how can you log out?", ToastLength.Short).Show ();
			}
			Gateway.ResetAccessToken ();
			UnlockLoginUI ();
		}
	}
}