text:
package org.elasticsearch.monitor.os;

import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.io.stream.Streamable;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;

import java.io.IOException;
import java.util.Arrays;

/**
 *
 */
public class OsStats implements Streamable, ToXContent {

    long timestamp;

    Cpu cpu = null;

    Mem mem = null;

    Swap swap = null;

    OsStats() {
    }

    public long getTimestamp() {
        return timestamp;
    }

    public Cpu getCpu() { return cpu; }

    public Mem getMem() {
        return mem;
    }

    public Swap getSwap() {
        return swap;
    }

    static final class Fields {
        static final String OS = "os";
        static final String TIMESTAMP = "timestamp";
        static final String CPU = "cpu";
        static final String PERCENT = "percent";
        static final String LOAD_AVERAGE = "load_average";
        static final String LOAD_AVERAGE_1M = new String("1m");
        static final String LOAD_AVERAGE_5M = new String("5m");
        static final String LOAD_AVERAGE_15M = new String("15m");

        static final String MEM = "mem";
        static final String SWAP = "swap";
        static final String FREE = "free";
        static final String FREE_IN_BYTES = "free_in_bytes";
        static final String USED = "used";
        static final String USED_IN_BYTES = "used_in_bytes";
        static final String TOTAL = "total";
        static final String TOTAL_IN_BYTES = "total_in_bytes";

        static final String FREE_PERCENT = "free_percent";
        static final String USED_PERCENT = "used_percent";
    }

    @Override
    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
        builder.startObject(Fields.OS);
        builder.field(Fields.TIMESTAMP, getTimestamp());
        if (cpu != null) {
            builder.startObject(Fields.CPU);
            builder.field(Fields.PERCENT, cpu.getPercent());
            if (cpu.getLoadAverage() != null && Arrays.stream(cpu.getLoadAverage()).anyMatch(load -> load != -1)) {
                builder.startObject(Fields.LOAD_AVERAGE);
                if (cpu.getLoadAverage()[0] != -1) {
                    builder.field(Fields.LOAD_AVERAGE_1M, cpu.getLoadAverage()[0]);
                }
                if (cpu.getLoadAverage()[1] != -1) {
                    builder.field(Fields.LOAD_AVERAGE_5M, cpu.getLoadAverage()[1]);
                }
                if (cpu.getLoadAverage()[2] != -1) {
                    builder.field(Fields.LOAD_AVERAGE_15M, cpu.getLoadAverage()[2]);
                }
                builder.endObject();
            }
            builder.endObject();
        }

        if (mem != null) {
            builder.startObject(Fields.MEM);
            builder.byteSizeField(Fields.TOTAL_IN_BYTES, Fields.TOTAL, mem.getTotal());
            builder.byteSizeField(Fields.FREE_IN_BYTES, Fields.FREE, mem.getFree());
            builder.byteSizeField(Fields.USED_IN_BYTES, Fields.USED, mem.getUsed());

            builder.field(Fields.FREE_PERCENT, mem.getFreePercent());
            builder.field(Fields.USED_PERCENT, mem.getUsedPercent());

            builder.endObject();
        }

        if (swap != null) {
            builder.startObject(Fields.SWAP);
            builder.byteSizeField(Fields.TOTAL_IN_BYTES, Fields.TOTAL, swap.getTotal());
            builder.byteSizeField(Fields.FREE_IN_BYTES, Fields.FREE, swap.getFree());
            builder.byteSizeField(Fields.USED_IN_BYTES, Fields.USED, swap.getUsed());
            builder.endObject();
        }

        builder.endObject();
        return builder;
    }

    public static OsStats readOsStats(StreamInput in) throws IOException {
        OsStats stats = new OsStats();
        stats.readFrom(in);
        return stats;
    }

    @Override
    public void readFrom(StreamInput in) throws IOException {
        timestamp = in.readVLong();
        cpu = in.readOptionalStreamable(Cpu::new);
        if (in.readBoolean()) {
            mem = Mem.readMem(in);
        }
        if (in.readBoolean()) {
            swap = Swap.readSwap(in);
        }
    }

    @Override
    public void writeTo(StreamOutput out) throws IOException {
        out.writeVLong(timestamp);
        out.writeOptionalStreamable(cpu);
        if (mem == null) {
            out.writeBoolean(false);
        } else {
            out.writeBoolean(true);
            mem.writeTo(out);
        }
        if (swap == null) {
            out.writeBoolean(false);
        } else {
            out.writeBoolean(true);
            swap.writeTo(out);
        }
    }

    public static class Cpu implements Streamable {

        short percent = -1;
        double[] loadAverage = null;

        Cpu() {}

        public static Cpu readCpu(StreamInput in) throws IOException {
            Cpu cpu = new Cpu();
            cpu.readFrom(in);
            return cpu;
        }

        @Override
        public void readFrom(StreamInput in) throws IOException {
            percent = in.readShort();
            if (in.readBoolean()) {
                loadAverage = in.readDoubleArray();
            } else {
                loadAverage = null;
            }
        }

        @Override
        public void writeTo(StreamOutput out) throws IOException {
            out.writeShort(percent);
            if (loadAverage == null) {
                out.writeBoolean(false);
            } else {
                out.writeBoolean(true);
                out.writeDoubleArray(loadAverage);
            }
        }

        public short getPercent() {
            return percent;
        }

        public double[] getLoadAverage() {
            return loadAverage;
        }
    }

    public static class Swap implements Streamable {

        long total = -1;
        long free = -1;

        public ByteSizeValue getFree() {
            return new ByteSizeValue(free);
        }

        public ByteSizeValue getUsed() {
            return new ByteSizeValue(total - free);
        }

        public ByteSizeValue getTotal() {
            return new ByteSizeValue(total);
        }

        public static Swap readSwap(StreamInput in) throws IOException {
            Swap swap = new Swap();
            swap.readFrom(in);
            return swap;
        }

        @Override
        public void readFrom(StreamInput in) throws IOException {
            total = in.readLong();
            free = in.readLong();
        }

        @Override
        public void writeTo(StreamOutput out) throws IOException {
            out.writeLong(total);
            out.writeLong(free);
        }
    }

    public static class Mem implements Streamable {

        long total = -1;
        long free = -1;

        public static Mem readMem(StreamInput in) throws IOException {
            Mem mem = new Mem();
            mem.readFrom(in);
            return mem;
        }

        @Override
        public void readFrom(StreamInput in) throws IOException {
            total = in.readLong();
            free = in.readLong();
        }

        @Override
        public void writeTo(StreamOutput out) throws IOException {
            out.writeLong(total);
            out.writeLong(free);
        }

        public ByteSizeValue getTotal() {
            return new ByteSizeValue(total);
        }

        public ByteSizeValue getUsed() {
            return new ByteSizeValue(total - free);
        }

        public short getUsedPercent() {
            return calculatePercentage(getUsed().bytes(), getTotal().bytes());
        }

        public ByteSizeValue getFree() {
            return new ByteSizeValue(free);
        }

        public short getFreePercent() {
            return calculatePercentage(getFree().bytes(), getTotal().bytes());
        }
    }

    private static short calculatePercentage(long used, long max) {
        return max <= 0 ? 0 : (short) (Math.round((100d * used) / max));
    }
}


text:
require 'minitest_helper'

# BarkSpec is used to test the main client class
module BarkSpec
  describe 'Status', 'The Monit XML API Status Class' do
    it 'should initialize the correct type of client object' do
      client = Bark::Client.new
      expect(client.status).must_be_instance_of Bark::Status
    end

    it 'should retrieve the current status from Monit' do
      client = Bark::Client.new
      response = client.status.current
      expect(response[:monit].keys).must_equal [:server, :platform, :service]
    end
  end
end

redpajama_set_name: RedPajamaGithub


text:
#include <iostream>
#include "network/SocketClient.h"               // using the EBB library
#include "bus/I2CDevice.h"                      // I2CDevice class see CHP8
#include "sensor/ADXL345.h"                     // ADXL345 see CHP8
#include "sstream"                              // to format the string
#include <unistd.h>                             // for the usleep()
using namespace std;
using namespace exploringBB;

int main(int argc, char *argv[]){
   if(argc!=2){
      cout << "Usage:  accClient server_name" << endl;
      return 2;
   }
   cout << "Starting EBB ADXL345 Client Example" << endl;
   I2CDevice i2c(1,0x53);                       // the I2C device P9_19 P9_20
   ADXL345 sensor(&i2c);                        // pass device to ADXL const.
   sensor.setResolution(ADXL345::NORMAL);       // regular resolution
   sensor.setRange(ADXL345::PLUSMINUS_4_G);     // regular +/- 2G

   SocketClient sc(argv[1], 4444);              // server addr and port number
   sc.connectToServer();                        // connect to the server
   for(int i=0; i<1000; i++){                   // going to send 1000 samples
      stringstream ss;                          // use a stringstream for msg.
      sensor.readSensorState();                 // update the sensor state
      float pitch = sensor.getPitch();          // get pitch and roll
      float roll = sensor.getRoll();            // structure as XML string
      ss << "<sample><acc><pitch>" << pitch << "</pitch>";
      ss << "<roll>" << roll << "</roll></acc></sample>      ";
      cout << ss.str() << '\xd';                // print to output on one line
      cout.flush();                             // flush to update the display
      sc.send(ss.str());                        // send the same str to server
      usleep(50000);                            //50ms between samples
   }
   cout << "End of EBB Client Example" << endl;
}


text:
package kafka.coordinator.group

import java.io.PrintStream
import java.nio.ByteBuffer
import java.nio.charset.StandardCharsets
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.locks.ReentrantLock

import com.yammer.metrics.core.Gauge
import kafka.api.{ApiVersion, KAFKA_0_10_1_IV0}
import kafka.common.{KafkaException, MessageFormatter, OffsetAndMetadata}
import kafka.metrics.KafkaMetricsGroup
import kafka.server.ReplicaManager
import kafka.utils.CoreUtils.inLock
import kafka.utils._
import kafka.zk.KafkaZkClient
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.protocol.types.Type._
import org.apache.kafka.common.protocol.types._
import org.apache.kafka.common.record._
import org.apache.kafka.common.requests.{IsolationLevel, OffsetFetchResponse}
import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
import org.apache.kafka.common.utils.{Time, Utils}

import scala.collection.JavaConverters._
import scala.collection._
import scala.collection.mutable.ListBuffer

class GroupMetadataManager(brokerId: Int,
                           interBrokerProtocolVersion: ApiVersion,
                           config: OffsetConfig,
                           replicaManager: ReplicaManager,
                           zkClient: KafkaZkClient,
                           time: Time) extends Logging with KafkaMetricsGroup {

  private val compressionType: CompressionType = CompressionType.forId(config.offsetsTopicCompressionCodec.codec)

  private val groupMetadataCache = new Pool[String, GroupMetadata]

  /* lock protecting access to loading and owned partition sets */
  private val partitionLock = new ReentrantLock()

  /* partitions of consumer groups that are being loaded, its lock should be always called BEFORE the group lock if needed */
  private val loadingPartitions: mutable.Set[Int] = mutable.Set()

  /* partitions of consumer groups that are assigned, using the same loading partition lock */
  private val ownedPartitions: mutable.Set[Int] = mutable.Set()

  /* shutting down flag */
  private val shuttingDown = new AtomicBoolean(false)

  /* number of partitions for the consumer metadata topic */
  private val groupMetadataTopicPartitionCount = getGroupMetadataTopicPartitionCount

  /* single-thread scheduler to handle offset/group metadata cache loading and unloading */
  private val scheduler = new KafkaScheduler(threads = 1, threadNamePrefix = "group-metadata-manager-")

  /* The groups with open transactional offsets commits per producer. We need this because when the commit or abort
   * marker comes in for a transaction, it is for a particular partition on the offsets topic and a particular producerId.
   * We use this structure to quickly find the groups which need to be updated by the commit/abort marker. */
  private val openGroupsForProducer = mutable.HashMap[Long, mutable.Set[String]]()

  this.logIdent = s"[GroupMetadataManager brokerId=$brokerId] "

  private def recreateGauge[T](name: String, gauge: Gauge[T]): Gauge[T] = {
    removeMetric(name)
    newGauge(name, gauge)
  }

  recreateGauge("NumOffsets",
    new Gauge[Int] {
      def value = groupMetadataCache.values.map(group => {
        group.inLock { group.numOffsets }
      }).sum
    })

  recreateGauge("NumGroups",
    new Gauge[Int] {
      def value = groupMetadataCache.size
    })

  recreateGauge("NumGroupsPreparingRebalance",
    new Gauge[Int] {
      def value(): Int = groupMetadataCache.values.count(group => {
        group synchronized { group.is(PreparingRebalance) }
      })
    })

  recreateGauge("NumGroupsCompletingRebalance",
    new Gauge[Int] {
      def value(): Int = groupMetadataCache.values.count(group => {
        group synchronized { group.is(CompletingRebalance) }
      })
    })

  recreateGauge("NumGroupsStable",
    new Gauge[Int] {
      def value(): Int = groupMetadataCache.values.count(group => {
        group synchronized { group.is(Stable) }
      })
    })

  recreateGauge("NumGroupsDead",
    new Gauge[Int] {
      def value(): Int = groupMetadataCache.values.count(group => {
        group synchronized { group.is(Dead) }
      })
    })

  recreateGauge("NumGroupsEmpty",
    new Gauge[Int] {
      def value(): Int = groupMetadataCache.values.count(group => {
        group synchronized { group.is(Empty) }
      })
    })

  def enableMetadataExpiration() {
    scheduler.startup()

    scheduler.schedule(name = "delete-expired-group-metadata",
      fun = cleanupGroupMetadata,
      period = config.offsetsRetentionCheckIntervalMs,
      unit = TimeUnit.MILLISECONDS)
  }

  def currentGroups: Iterable[GroupMetadata] = groupMetadataCache.values

  def isPartitionOwned(partition: Int) = inLock(partitionLock) { ownedPartitions.contains(partition) }

  def isPartitionLoading(partition: Int) = inLock(partitionLock) { loadingPartitions.contains(partition) }

  def partitionFor(groupId: String): Int = Utils.abs(groupId.hashCode) % groupMetadataTopicPartitionCount

  def isGroupLocal(groupId: String): Boolean = isPartitionOwned(partitionFor(groupId))

  def isGroupLoading(groupId: String): Boolean = isPartitionLoading(partitionFor(groupId))

  def isLoading(): Boolean = inLock(partitionLock) { loadingPartitions.nonEmpty }

  // return true iff group is owned and the group doesn't exist
  def groupNotExists(groupId: String) = inLock(partitionLock) {
    isGroupLocal(groupId) && getGroup(groupId).forall { group =>
      group.inLock(group.is(Dead))
    }
  }

  // visible for testing
  private[group] def isGroupOpenForProducer(producerId: Long, groupId: String) = openGroupsForProducer.get(producerId) match {
    case Some(groups) =>
      groups.contains(groupId)
    case None =>
      false
  }
  /**
   * Get the group associated with the given groupId, or null if not found
   */
  def getGroup(groupId: String): Option[GroupMetadata] = {
    Option(groupMetadataCache.get(groupId))
  }

  /**
   * Add a group or get the group associated with the given groupId if it already exists
   */
  def addGroup(group: GroupMetadata): GroupMetadata = {
    val currentGroup = groupMetadataCache.putIfNotExists(group.groupId, group)
    if (currentGroup != null) {
      currentGroup
    } else {
      group
    }
  }

  def storeGroup(group: GroupMetadata,
                 groupAssignment: Map[String, Array[Byte]],
                 responseCallback: Errors => Unit): Unit = {
    getMagic(partitionFor(group.groupId)) match {
      case Some(magicValue) =>
        val groupMetadataValueVersion = {
          if (interBrokerProtocolVersion < KAFKA_0_10_1_IV0)
            0.toShort
          else
            GroupMetadataManager.CURRENT_GROUP_VALUE_SCHEMA_VERSION
        }

        // We always use CREATE_TIME, like the producer. The conversion to LOG_APPEND_TIME (if necessary) happens automatically.
        val timestampType = TimestampType.CREATE_TIME
        val timestamp = time.milliseconds()
        val key = GroupMetadataManager.groupMetadataKey(group.groupId)
        val value = GroupMetadataManager.groupMetadataValue(group, groupAssignment, version = groupMetadataValueVersion)

        val records = {
          val buffer = ByteBuffer.allocate(AbstractRecords.estimateSizeInBytes(magicValue, compressionType,
            Seq(new SimpleRecord(timestamp, key, value)).asJava))
          val builder = MemoryRecords.builder(buffer, magicValue, compressionType, timestampType, 0L)
          builder.append(timestamp, key, value)
          builder.build()
        }

        val groupMetadataPartition = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, partitionFor(group.groupId))
        val groupMetadataRecords = Map(groupMetadataPartition -> records)
        val generationId = group.generationId

        // set the callback function to insert the created group into cache after log append completed
        def putCacheCallback(responseStatus: Map[TopicPartition, PartitionResponse]) {
          // the append response should only contain the topics partition
          if (responseStatus.size != 1 || !responseStatus.contains(groupMetadataPartition))
            throw new IllegalStateException("Append status %s should only have one partition %s"
              .format(responseStatus, groupMetadataPartition))

          // construct the error status in the propagated assignment response in the cache
          val status = responseStatus(groupMetadataPartition)

          val responseError = if (status.error == Errors.NONE) {
            Errors.NONE
          } else {
            debug(s"Metadata from group ${group.groupId} with generation $generationId failed when appending to log " +
              s"due to ${status.error.exceptionName}")

            // transform the log append error code to the corresponding the commit status error code
            status.error match {
              case Errors.UNKNOWN_TOPIC_OR_PARTITION
                   | Errors.NOT_ENOUGH_REPLICAS
                   | Errors.NOT_ENOUGH_REPLICAS_AFTER_APPEND =>
                Errors.COORDINATOR_NOT_AVAILABLE

              case Errors.NOT_LEADER_FOR_PARTITION
                   | Errors.KAFKA_STORAGE_ERROR =>
                Errors.NOT_COORDINATOR

              case Errors.REQUEST_TIMED_OUT =>
                Errors.REBALANCE_IN_PROGRESS

              case Errors.MESSAGE_TOO_LARGE
                   | Errors.RECORD_LIST_TOO_LARGE
                   | Errors.INVALID_FETCH_SIZE =>

                error(s"Appending metadata message for group ${group.groupId} generation $generationId failed due to " +
                  s"${status.error.exceptionName}, returning UNKNOWN error code to the client")

                Errors.UNKNOWN_SERVER_ERROR

              case other =>
                error(s"Appending metadata message for group ${group.groupId} generation $generationId failed " +
                  s"due to unexpected error: ${status.error.exceptionName}")

                other
            }
          }

          responseCallback(responseError)
        }
        appendForGroup(group, groupMetadataRecords, putCacheCallback)

      case None =>
        responseCallback(Errors.NOT_COORDINATOR)
        None
    }
  }

  private def appendForGroup(group: GroupMetadata,
                             records: Map[TopicPartition, MemoryRecords],
                             callback: Map[TopicPartition, PartitionResponse] => Unit): Unit = {
    // call replica manager to append the group message
    replicaManager.appendRecords(
      timeout = config.offsetCommitTimeoutMs.toLong,
      requiredAcks = config.offsetCommitRequiredAcks,
      internalTopicsAllowed = true,
      isFromClient = false,
      entriesPerPartition = records,
      delayedProduceLock = Some(group.lock),
      responseCallback = callback)
  }

  /**
   * Store offsets by appending it to the replicated log and then inserting to cache
   */
  def storeOffsets(group: GroupMetadata,
                   consumerId: String,
                   offsetMetadata: immutable.Map[TopicPartition, OffsetAndMetadata],
                   responseCallback: immutable.Map[TopicPartition, Errors] => Unit,
                   producerId: Long = RecordBatch.NO_PRODUCER_ID,
                   producerEpoch: Short = RecordBatch.NO_PRODUCER_EPOCH): Unit = {
    // first filter out partitions with offset metadata size exceeding limit
    val filteredOffsetMetadata = offsetMetadata.filter { case (_, offsetAndMetadata) =>
      validateOffsetMetadataLength(offsetAndMetadata.metadata)
    }

    group.inLock {
      if (!group.hasReceivedConsistentOffsetCommits)
        warn(s"group: ${group.groupId} with leader: ${group.leaderOrNull} has received offset commits from consumers as well " +
          s"as transactional producers. Mixing both types of offset commits will generally result in surprises and " +
          s"should be avoided.")
    }

    val isTxnOffsetCommit = producerId != RecordBatch.NO_PRODUCER_ID
    // construct the message set to append
    if (filteredOffsetMetadata.isEmpty) {
      // compute the final error codes for the commit response
      val commitStatus = offsetMetadata.mapValues(_ => Errors.OFFSET_METADATA_TOO_LARGE)
      responseCallback(commitStatus)
      None
    } else {
      getMagic(partitionFor(group.groupId)) match {
        case Some(magicValue) =>
          // We always use CREATE_TIME, like the producer. The conversion to LOG_APPEND_TIME (if necessary) happens automatically.
          val timestampType = TimestampType.CREATE_TIME
          val timestamp = time.milliseconds()

          val records = filteredOffsetMetadata.map { case (topicPartition, offsetAndMetadata) =>
            val key = GroupMetadataManager.offsetCommitKey(group.groupId, topicPartition)
            val value = GroupMetadataManager.offsetCommitValue(offsetAndMetadata)
            new SimpleRecord(timestamp, key, value)
          }
          val offsetTopicPartition = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, partitionFor(group.groupId))
          val buffer = ByteBuffer.allocate(AbstractRecords.estimateSizeInBytes(magicValue, compressionType, records.asJava))

          if (isTxnOffsetCommit && magicValue < RecordBatch.MAGIC_VALUE_V2)
            throw Errors.UNSUPPORTED_FOR_MESSAGE_FORMAT.exception("Attempting to make a transaction offset commit with an invalid magic: " + magicValue)

          val builder = MemoryRecords.builder(buffer, magicValue, compressionType, timestampType, 0L, time.milliseconds(),
            producerId, producerEpoch, 0, isTxnOffsetCommit, RecordBatch.NO_PARTITION_LEADER_EPOCH)

          records.foreach(builder.append)
          val entries = Map(offsetTopicPartition -> builder.build())

          // set the callback function to insert offsets into cache after log append completed
          def putCacheCallback(responseStatus: Map[TopicPartition, PartitionResponse]) {
            // the append response should only contain the topics partition
            if (responseStatus.size != 1 || !responseStatus.contains(offsetTopicPartition))
              throw new IllegalStateException("Append status %s should only have one partition %s"
                .format(responseStatus, offsetTopicPartition))

            // construct the commit response status and insert
            // the offset and metadata to cache if the append status has no error
            val status = responseStatus(offsetTopicPartition)

            val responseError = group.inLock {
              if (status.error == Errors.NONE) {
                if (!group.is(Dead)) {
                  filteredOffsetMetadata.foreach { case (topicPartition, offsetAndMetadata) =>
                    if (isTxnOffsetCommit)
                      group.onTxnOffsetCommitAppend(producerId, topicPartition, CommitRecordMetadataAndOffset(Some(status.baseOffset), offsetAndMetadata))
                    else
                      group.onOffsetCommitAppend(topicPartition, CommitRecordMetadataAndOffset(Some(status.baseOffset), offsetAndMetadata))
                  }
                }
                Errors.NONE
              } else {
                if (!group.is(Dead)) {
                  if (!group.hasPendingOffsetCommitsFromProducer(producerId))
                    removeProducerGroup(producerId, group.groupId)
                  filteredOffsetMetadata.foreach { case (topicPartition, offsetAndMetadata) =>
                    if (isTxnOffsetCommit)
                      group.failPendingTxnOffsetCommit(producerId, topicPartition)
                    else
                      group.failPendingOffsetWrite(topicPartition, offsetAndMetadata)
                  }
                }

                debug(s"Offset commit $filteredOffsetMetadata from group ${group.groupId}, consumer $consumerId " +
                  s"with generation ${group.generationId} failed when appending to log due to ${status.error.exceptionName}")

                // transform the log append error code to the corresponding the commit status error code
                status.error match {
                  case Errors.UNKNOWN_TOPIC_OR_PARTITION
                       | Errors.NOT_ENOUGH_REPLICAS
                       | Errors.NOT_ENOUGH_REPLICAS_AFTER_APPEND =>
                    Errors.COORDINATOR_NOT_AVAILABLE

                  case Errors.NOT_LEADER_FOR_PARTITION
                       | Errors.KAFKA_STORAGE_ERROR =>
                    Errors.NOT_COORDINATOR

                  case Errors.MESSAGE_TOO_LARGE
                       | Errors.RECORD_LIST_TOO_LARGE
                       | Errors.INVALID_FETCH_SIZE =>
                    Errors.INVALID_COMMIT_OFFSET_SIZE

                  case other => other
                }
              }
            }

            // compute the final error codes for the commit response
            val commitStatus = offsetMetadata.map { case (topicPartition, offsetAndMetadata) =>
              if (validateOffsetMetadataLength(offsetAndMetadata.metadata))
                (topicPartition, responseError)
              else
                (topicPartition, Errors.OFFSET_METADATA_TOO_LARGE)
            }

            // finally trigger the callback logic passed from the API layer
            responseCallback(commitStatus)
          }

          if (isTxnOffsetCommit) {
            group.inLock {
              addProducerGroup(producerId, group.groupId)
              group.prepareTxnOffsetCommit(producerId, offsetMetadata)
            }
          } else {
            group.inLock {
              group.prepareOffsetCommit(offsetMetadata)
            }
          }

          appendForGroup(group, entries, putCacheCallback)

        case None =>
          val commitStatus = offsetMetadata.map { case (topicPartition, _) =>
            (topicPartition, Errors.NOT_COORDINATOR)
          }
          responseCallback(commitStatus)
          None
      }
    }
  }

  /**
   * The most important guarantee that this API provides is that it should never return a stale offset. i.e., it either
   * returns the current offset or it begins to sync the cache from the log (and returns an error code).
   */
  def getOffsets(groupId: String, topicPartitionsOpt: Option[Seq[TopicPartition]]): Map[TopicPartition, OffsetFetchResponse.PartitionData] = {
    trace("Getting offsets of %s for group %s.".format(topicPartitionsOpt.getOrElse("all partitions"), groupId))
    val group = groupMetadataCache.get(groupId)
    if (group == null) {
      topicPartitionsOpt.getOrElse(Seq.empty[TopicPartition]).map { topicPartition =>
        (topicPartition, new OffsetFetchResponse.PartitionData(OffsetFetchResponse.INVALID_OFFSET, "", Errors.NONE))
      }.toMap
    } else {
      group.inLock {
        if (group.is(Dead)) {
          topicPartitionsOpt.getOrElse(Seq.empty[TopicPartition]).map { topicPartition =>
            (topicPartition, new OffsetFetchResponse.PartitionData(OffsetFetchResponse.INVALID_OFFSET, "", Errors.NONE))
          }.toMap
        } else {
          topicPartitionsOpt match {
            case None =>
              // Return offsets for all partitions owned by this consumer group. (this only applies to consumers
              // that commit offsets to Kafka.)
              group.allOffsets.map { case (topicPartition, offsetAndMetadata) =>
                topicPartition -> new OffsetFetchResponse.PartitionData(offsetAndMetadata.offset, offsetAndMetadata.metadata, Errors.NONE)
              }

            case Some(topicPartitions) =>
              topicPartitionsOpt.getOrElse(Seq.empty[TopicPartition]).map { topicPartition =>
                val partitionData = group.offset(topicPartition) match {
                  case None =>
                    new OffsetFetchResponse.PartitionData(OffsetFetchResponse.INVALID_OFFSET, "", Errors.NONE)
                  case Some(offsetAndMetadata) =>
                    new OffsetFetchResponse.PartitionData(offsetAndMetadata.offset, offsetAndMetadata.metadata, Errors.NONE)
                }
                topicPartition -> partitionData
              }.toMap
          }
        }
      }
    }
  }

  /**
   * Asynchronously read the partition from the offsets topic and populate the cache
   */
  def loadGroupsForPartition(offsetsPartition: Int, onGroupLoaded: GroupMetadata => Unit) {
    val topicPartition = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, offsetsPartition)
    info(s"Scheduling loading of offsets and group metadata from $topicPartition")
    scheduler.schedule(topicPartition.toString, doLoadGroupsAndOffsets)

    def doLoadGroupsAndOffsets() {
      inLock(partitionLock) {
        if (loadingPartitions.contains(offsetsPartition)) {
          info(s"Offset load from $topicPartition already in progress.")
          return
        } else {
          loadingPartitions.add(offsetsPartition)
        }
      }

      try {
        val startMs = time.milliseconds()
        loadGroupsAndOffsets(topicPartition, onGroupLoaded)
        info(s"Finished loading offsets and group metadata from $topicPartition in ${time.milliseconds() - startMs} milliseconds.")
      } catch {
        case t: Throwable => error(s"Error loading offsets from $topicPartition", t)
      } finally {
        inLock(partitionLock) {
          ownedPartitions.add(offsetsPartition)
          loadingPartitions.remove(offsetsPartition)
        }
      }
    }
  }

  private[group] def loadGroupsAndOffsets(topicPartition: TopicPartition, onGroupLoaded: GroupMetadata => Unit) {
    def highWaterMark = replicaManager.getLogEndOffset(topicPartition).getOrElse(-1L)

    replicaManager.getLog(topicPartition) match {
      case None =>
        warn(s"Attempted to load offsets and group metadata from $topicPartition, but found no log")

      case Some(log) =>
        var currOffset = log.logStartOffset
        lazy val buffer = ByteBuffer.allocate(config.loadBufferSize)

        // loop breaks if leader changes at any time during the load, since getHighWatermark is -1
        val loadedOffsets = mutable.Map[GroupTopicPartition, CommitRecordMetadataAndOffset]()
        val pendingOffsets = mutable.Map[Long, mutable.Map[GroupTopicPartition, CommitRecordMetadataAndOffset]]()
        val loadedGroups = mutable.Map[String, GroupMetadata]()
        val removedGroups = mutable.Set[String]()

        while (currOffset < highWaterMark && !shuttingDown.get()) {
          val fetchDataInfo = log.read(currOffset, config.loadBufferSize, maxOffset = None,
            minOneMessage = true, isolationLevel = IsolationLevel.READ_UNCOMMITTED)
          val memRecords = fetchDataInfo.records match {
            case records: MemoryRecords => records
            case fileRecords: FileRecords =>
              buffer.clear()
              val bufferRead = fileRecords.readInto(buffer, 0)
              MemoryRecords.readableRecords(bufferRead)
          }

          memRecords.batches.asScala.foreach { batch =>
            val isTxnOffsetCommit = batch.isTransactional
            if (batch.isControlBatch) {
              val record = batch.iterator.next()
              val controlRecord = ControlRecordType.parse(record.key)
              if (controlRecord == ControlRecordType.COMMIT) {
                pendingOffsets.getOrElse(batch.producerId, mutable.Map[GroupTopicPartition, CommitRecordMetadataAndOffset]())
                  .foreach {
                    case (groupTopicPartition, commitRecordMetadataAndOffset) =>
                      if (!loadedOffsets.contains(groupTopicPartition) || loadedOffsets(groupTopicPartition).olderThan(commitRecordMetadataAndOffset))
                        loadedOffsets.put(groupTopicPartition, commitRecordMetadataAndOffset)
                  }
              }
              pendingOffsets.remove(batch.producerId)
            } else {
              for (record <- batch.asScala) {
                require(record.hasKey, "Group metadata/offset entry key should not be null")
                GroupMetadataManager.readMessageKey(record.key) match {

                  case offsetKey: OffsetKey =>
                    if (isTxnOffsetCommit && !pendingOffsets.contains(batch.producerId))
                      pendingOffsets.put(batch.producerId, mutable.Map[GroupTopicPartition, CommitRecordMetadataAndOffset]())

                    // load offset
                    val groupTopicPartition = offsetKey.key
                    if (!record.hasValue) {
                      if (isTxnOffsetCommit)
                        pendingOffsets(batch.producerId).remove(groupTopicPartition)
                      else
                        loadedOffsets.remove(groupTopicPartition)
                    } else {
                      val offsetAndMetadata = GroupMetadataManager.readOffsetMessageValue(record.value)
                      if (isTxnOffsetCommit)
                        pendingOffsets(batch.producerId).put(groupTopicPartition, CommitRecordMetadataAndOffset(Some(batch.baseOffset), offsetAndMetadata))
                      else
                        loadedOffsets.put(groupTopicPartition, CommitRecordMetadataAndOffset(Some(batch.baseOffset), offsetAndMetadata))
                    }

                  case groupMetadataKey: GroupMetadataKey =>
                    // load group metadata
                    val groupId = groupMetadataKey.key
                    val groupMetadata = GroupMetadataManager.readGroupMessageValue(groupId, record.value)
                    if (groupMetadata != null) {
                      removedGroups.remove(groupId)
                      loadedGroups.put(groupId, groupMetadata)
                    } else {
                      loadedGroups.remove(groupId)
                      removedGroups.add(groupId)
                    }

                  case unknownKey =>
                    throw new IllegalStateException(s"Unexpected message key $unknownKey while loading offsets and group metadata")
                }
              }
            }
            currOffset = batch.nextOffset
          }
        }

        val (groupOffsets, emptyGroupOffsets) = loadedOffsets
          .groupBy(_._1.group)
          .mapValues(_.map { case (groupTopicPartition, offset) => (groupTopicPartition.topicPartition, offset) })
          .partition { case (group, _) => loadedGroups.contains(group) }

        val pendingOffsetsByGroup = mutable.Map[String, mutable.Map[Long, mutable.Map[TopicPartition, CommitRecordMetadataAndOffset]]]()
        pendingOffsets.foreach { case (producerId, producerOffsets) =>
          producerOffsets.keySet.map(_.group).foreach(addProducerGroup(producerId, _))
          producerOffsets
            .groupBy(_._1.group)
            .mapValues(_.map { case (groupTopicPartition, offset) => (groupTopicPartition.topicPartition, offset)})
            .foreach { case (group, offsets) =>
              val groupPendingOffsets = pendingOffsetsByGroup.getOrElseUpdate(group, mutable.Map.empty[Long, mutable.Map[TopicPartition, CommitRecordMetadataAndOffset]])
              val groupProducerOffsets = groupPendingOffsets.getOrElseUpdate(producerId, mutable.Map.empty[TopicPartition, CommitRecordMetadataAndOffset])
              groupProducerOffsets ++= offsets
            }
        }

        val (pendingGroupOffsets, pendingEmptyGroupOffsets) = pendingOffsetsByGroup
          .partition { case (group, _) => loadedGroups.contains(group)}

        loadedGroups.values.foreach { group =>
          val offsets = groupOffsets.getOrElse(group.groupId, Map.empty[TopicPartition, CommitRecordMetadataAndOffset])
          val pendingOffsets = pendingGroupOffsets.getOrElse(group.groupId, Map.empty[Long, mutable.Map[TopicPartition, CommitRecordMetadataAndOffset]])
          debug(s"Loaded group metadata $group with offsets $offsets and pending offsets $pendingOffsets")
          loadGroup(group, offsets, pendingOffsets)
          onGroupLoaded(group)
        }

        // load groups which store offsets in kafka, but which have no active members and thus no group
        // metadata stored in the log
        (emptyGroupOffsets.keySet ++ pendingEmptyGroupOffsets.keySet).foreach { groupId =>
          val group = new GroupMetadata(groupId, initialState = Empty)
          val offsets = emptyGroupOffsets.getOrElse(groupId, Map.empty[TopicPartition, CommitRecordMetadataAndOffset])
          val pendingOffsets = pendingEmptyGroupOffsets.getOrElse(groupId, Map.empty[Long, mutable.Map[TopicPartition, CommitRecordMetadataAndOffset]])
          debug(s"Loaded group metadata $group with offsets $offsets and pending offsets $pendingOffsets")
          loadGroup(group, offsets, pendingOffsets)
          onGroupLoaded(group)
        }

        removedGroups.foreach { groupId =>
          // if the cache already contains a group which should be removed, raise an error. Note that it
          // is possible (however unlikely) for a consumer group to be removed, and then to be used only for
          // offset storage (i.e. by "simple" consumers)
          if (groupMetadataCache.contains(groupId) && !emptyGroupOffsets.contains(groupId))
            throw new IllegalStateException(s"Unexpected unload of active group $groupId while " +
              s"loading partition $topicPartition")
        }

    }
  }

  private def loadGroup(group: GroupMetadata, offsets: Map[TopicPartition, CommitRecordMetadataAndOffset],
                        pendingTransactionalOffsets: Map[Long, mutable.Map[TopicPartition, CommitRecordMetadataAndOffset]]): Unit = {
    // offsets are initialized prior to loading the group into the cache to ensure that clients see a consistent
    // view of the group's offsets
    val loadedOffsets = offsets.mapValues { case CommitRecordMetadataAndOffset(commitRecordOffset, offsetAndMetadata) =>
      // special handling for version 0:
      // set the expiration time stamp as commit time stamp + server default retention time
      val updatedOffsetAndMetadata =
        if (offsetAndMetadata.expireTimestamp == org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_TIMESTAMP)
        offsetAndMetadata.copy(expireTimestamp = offsetAndMetadata.commitTimestamp + config.offsetsRetentionMs)
      else
        offsetAndMetadata
      CommitRecordMetadataAndOffset(commitRecordOffset, updatedOffsetAndMetadata)
    }
    trace(s"Initialized offsets $loadedOffsets for group ${group.groupId}")
    group.initializeOffsets(loadedOffsets, pendingTransactionalOffsets.toMap)

    val currentGroup = addGroup(group)
    if (group != currentGroup)
      debug(s"Attempt to load group ${group.groupId} from log with generation ${group.generationId} failed " +
        s"because there is already a cached group with generation ${currentGroup.generationId}")
  }

  /**
   * When this broker becomes a follower for an offsets topic partition clear out the cache for groups that belong to
   * that partition.
   *
   * @param offsetsPartition Groups belonging to this partition of the offsets topic will be deleted from the cache.
   */
  def removeGroupsForPartition(offsetsPartition: Int,
                               onGroupUnloaded: GroupMetadata => Unit) {
    val topicPartition = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, offsetsPartition)
    info(s"Scheduling unloading of offsets and group metadata from $topicPartition")
    scheduler.schedule(topicPartition.toString, removeGroupsAndOffsets)

    def removeGroupsAndOffsets() {
      var numOffsetsRemoved = 0
      var numGroupsRemoved = 0

      inLock(partitionLock) {
        // we need to guard the group removal in cache in the loading partition lock
        // to prevent coordinator's check-and-get-group race condition
        ownedPartitions.remove(offsetsPartition)

        for (group <- groupMetadataCache.values) {
          if (partitionFor(group.groupId) == offsetsPartition) {
            onGroupUnloaded(group)
            groupMetadataCache.remove(group.groupId, group)
            removeGroupFromAllProducers(group.groupId)
            numGroupsRemoved += 1
            numOffsetsRemoved += group.numOffsets
          }
        }
      }

      info(s"Finished unloading $topicPartition. Removed $numOffsetsRemoved cached offsets " +
        s"and $numGroupsRemoved cached groups.")
    }
  }

  // visible for testing
  private[group] def cleanupGroupMetadata(): Unit = {
    cleanupGroupMetadata(None, groupMetadataCache.values, time.milliseconds())
  }

  def cleanupGroupMetadata(deletedTopicPartitions: Option[Seq[TopicPartition]],
                           groups: Iterable[GroupMetadata],
                           startMs: Long) {
    var offsetsRemoved = 0

    groups.foreach { group =>
      val groupId = group.groupId
      val (removedOffsets, groupIsDead, generation) = group.inLock {
        val removedOffsets = deletedTopicPartitions match {
          case Some(topicPartitions) => group.removeOffsets(topicPartitions)
          case None => group.removeExpiredOffsets(startMs)
        }

        if (group.is(Empty) && !group.hasOffsets) {
          info(s"Group $groupId transitioned to Dead in generation ${group.generationId}")
          group.transitionTo(Dead)
        }
        (removedOffsets, group.is(Dead), group.generationId)
      }

      val offsetsPartition = partitionFor(groupId)
      val appendPartition = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, offsetsPartition)
      getMagic(offsetsPartition) match {
        case Some(magicValue) =>
          // We always use CREATE_TIME, like the producer. The conversion to LOG_APPEND_TIME (if necessary) happens automatically.
          val timestampType = TimestampType.CREATE_TIME
          val timestamp = time.milliseconds()

          replicaManager.nonOfflinePartition(appendPartition).foreach { partition =>
            val tombstones = ListBuffer.empty[SimpleRecord]
            removedOffsets.foreach { case (topicPartition, offsetAndMetadata) =>
              trace(s"Removing expired/deleted offset and metadata for $groupId, $topicPartition: $offsetAndMetadata")
              val commitKey = GroupMetadataManager.offsetCommitKey(groupId, topicPartition)
              tombstones += new SimpleRecord(timestamp, commitKey, null)
            }
            trace(s"Marked ${removedOffsets.size} offsets in $appendPartition for deletion.")

            // We avoid writing the tombstone when the generationId is 0, since this group is only using
            // Kafka for offset storage.
            if (groupIsDead && groupMetadataCache.remove(groupId, group) && generation > 0) {
              // Append the tombstone messages to the partition. It is okay if the replicas don't receive these (say,
              // if we crash or leaders move) since the new leaders will still expire the consumers with heartbeat and
              // retry removing this group.
              val groupMetadataKey = GroupMetadataManager.groupMetadataKey(group.groupId)
              tombstones += new SimpleRecord(timestamp, groupMetadataKey, null)
              trace(s"Group $groupId removed from the metadata cache and marked for deletion in $appendPartition.")
            }

            if (tombstones.nonEmpty) {
              try {
                // do not need to require acks since even if the tombstone is lost,
                // it will be appended again in the next purge cycle
                val records = MemoryRecords.withRecords(magicValue, 0L, compressionType, timestampType, tombstones: _*)
                partition.appendRecordsToLeader(records, isFromClient = false, requiredAcks = 0)

                offsetsRemoved += removedOffsets.size
                trace(s"Successfully appended ${tombstones.size} tombstones to $appendPartition for expired/deleted " +
                  s"offsets and/or metadata for group $groupId")
              } catch {
                case t: Throwable =>
                  error(s"Failed to append ${tombstones.size} tombstones to $appendPartition for expired/deleted " +
                    s"offsets and/or metadata for group $groupId.", t)
                // ignore and continue
              }
            }
          }

        case None =>
          info(s"BrokerId $brokerId is no longer a coordinator for the group $groupId. Proceeding cleanup for other alive groups")
      }
    }

    info(s"Removed $offsetsRemoved expired offsets in ${time.milliseconds() - startMs} milliseconds.")
  }

  def handleTxnCompletion(producerId: Long, completedPartitions: Set[Int], isCommit: Boolean) {
    val pendingGroups = groupsBelongingToPartitions(producerId, completedPartitions)
    pendingGroups.foreach { case (groupId) =>
      getGroup(groupId) match {
        case Some(group) => group.inLock {
          if (!group.is(Dead)) {
            group.completePendingTxnOffsetCommit(producerId, isCommit)
            removeProducerGroup(producerId, groupId)
          }
       }
        case _ =>
          info(s"Group $groupId has moved away from $brokerId after transaction marker was written but before the " +
            s"cache was updated. The cache on the new group owner will be updated instead.")
      }
    }
  }

  private def addProducerGroup(producerId: Long, groupId: String) = openGroupsForProducer synchronized {
    openGroupsForProducer.getOrElseUpdate(producerId, mutable.Set.empty[String]).add(groupId)
  }

  private def removeProducerGroup(producerId: Long, groupId: String) = openGroupsForProducer synchronized {
    openGroupsForProducer.getOrElseUpdate(producerId, mutable.Set.empty[String]).remove(groupId)
    if (openGroupsForProducer(producerId).isEmpty)
      openGroupsForProducer.remove(producerId)
  }

  private def groupsBelongingToPartitions(producerId: Long, partitions: Set[Int]) = openGroupsForProducer synchronized {
    val (ownedGroups, _) = openGroupsForProducer.getOrElse(producerId, mutable.Set.empty[String])
      .partition { case (group) => partitions.contains(partitionFor(group)) }
    ownedGroups
  }

  private def removeGroupFromAllProducers(groupId: String) = openGroupsForProducer synchronized {
    openGroupsForProducer.foreach { case (_, groups) =>
      groups.remove(groupId)
    }
  }

  /*
   * Check if the offset metadata length is valid
   */
  private def validateOffsetMetadataLength(metadata: String) : Boolean = {
    metadata == null || metadata.length() <= config.maxMetadataSize
  }

  def shutdown() {
    shuttingDown.set(true)
    if (scheduler.isStarted)
      scheduler.shutdown()

    // TODO: clear the caches
  }

  /**
   * Gets the partition count of the group metadata topic from ZooKeeper.
   * If the topic does not exist, the configured partition count is returned.
   */
  private def getGroupMetadataTopicPartitionCount: Int = {
    zkClient.getTopicPartitionCount(Topic.GROUP_METADATA_TOPIC_NAME).getOrElse(config.offsetsTopicNumPartitions)
  }

  /**
   * Check if the replica is local and return the message format version and timestamp
   *
   * @param   partition  Partition of GroupMetadataTopic
   * @return  Some(MessageFormatVersion) if replica is local, None otherwise
   */
  private def getMagic(partition: Int): Option[Byte] =
    replicaManager.getMagic(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, partition))

  /**
   * Add the partition into the owned list
   *
   * NOTE: this is for test only
   */
  def addPartitionOwnership(partition: Int) {
    inLock(partitionLock) {
      ownedPartitions.add(partition)
    }
  }
}

/**
 * Messages stored for the group topic has versions for both the key and value fields. Key
 * version is used to indicate the type of the message (also to differentiate different types
 * of messages from being compacted together if they have the same field values); and value
 * version is used to evolve the messages within their data types:
 *
 * key version 0:       group consumption offset
 *    -> value version 0:       [offset, metadata, timestamp]
 *
 * key version 1:       group consumption offset
 *    -> value version 1:       [offset, metadata, commit_timestamp, expire_timestamp]
 *
 * key version 2:       group metadata
 *     -> value version 0:       [protocol_type, generation, protocol, leader, members]
 */
object GroupMetadataManager {

  private val CURRENT_OFFSET_KEY_SCHEMA_VERSION = 1.toShort
  private val CURRENT_GROUP_KEY_SCHEMA_VERSION = 2.toShort

  private val OFFSET_COMMIT_KEY_SCHEMA = new Schema(new Field("group", STRING),
    new Field("topic", STRING),
    new Field("partition", INT32))
  private val OFFSET_KEY_GROUP_FIELD = OFFSET_COMMIT_KEY_SCHEMA.get("group")
  private val OFFSET_KEY_TOPIC_FIELD = OFFSET_COMMIT_KEY_SCHEMA.get("topic")
  private val OFFSET_KEY_PARTITION_FIELD = OFFSET_COMMIT_KEY_SCHEMA.get("partition")

  private val OFFSET_COMMIT_VALUE_SCHEMA_V0 = new Schema(new Field("offset", INT64),
    new Field("metadata", STRING, "Associated metadata.", ""),
    new Field("timestamp", INT64))
  private val OFFSET_VALUE_OFFSET_FIELD_V0 = OFFSET_COMMIT_VALUE_SCHEMA_V0.get("offset")
  private val OFFSET_VALUE_METADATA_FIELD_V0 = OFFSET_COMMIT_VALUE_SCHEMA_V0.get("metadata")
  private val OFFSET_VALUE_TIMESTAMP_FIELD_V0 = OFFSET_COMMIT_VALUE_SCHEMA_V0.get("timestamp")

  private val OFFSET_COMMIT_VALUE_SCHEMA_V1 = new Schema(new Field("offset", INT64),
    new Field("metadata", STRING, "Associated metadata.", ""),
    new Field("commit_timestamp", INT64),
    new Field("expire_timestamp", INT64))
  private val OFFSET_VALUE_OFFSET_FIELD_V1 = OFFSET_COMMIT_VALUE_SCHEMA_V1.get("offset")
  private val OFFSET_VALUE_METADATA_FIELD_V1 = OFFSET_COMMIT_VALUE_SCHEMA_V1.get("metadata")
  private val OFFSET_VALUE_COMMIT_TIMESTAMP_FIELD_V1 = OFFSET_COMMIT_VALUE_SCHEMA_V1.get("commit_timestamp")
  private val OFFSET_VALUE_EXPIRE_TIMESTAMP_FIELD_V1 = OFFSET_COMMIT_VALUE_SCHEMA_V1.get("expire_timestamp")

  private val GROUP_METADATA_KEY_SCHEMA = new Schema(new Field("group", STRING))
  private val GROUP_KEY_GROUP_FIELD = GROUP_METADATA_KEY_SCHEMA.get("group")

  private val MEMBER_ID_KEY = "member_id"
  private val CLIENT_ID_KEY = "client_id"
  private val CLIENT_HOST_KEY = "client_host"
  private val REBALANCE_TIMEOUT_KEY = "rebalance_timeout"
  private val SESSION_TIMEOUT_KEY = "session_timeout"
  private val SUBSCRIPTION_KEY = "subscription"
  private val ASSIGNMENT_KEY = "assignment"

  private val MEMBER_METADATA_V0 = new Schema(
    new Field(MEMBER_ID_KEY, STRING),
    new Field(CLIENT_ID_KEY, STRING),
    new Field(CLIENT_HOST_KEY, STRING),
    new Field(SESSION_TIMEOUT_KEY, INT32),
    new Field(SUBSCRIPTION_KEY, BYTES),
    new Field(ASSIGNMENT_KEY, BYTES))

  private val MEMBER_METADATA_V1 = new Schema(
    new Field(MEMBER_ID_KEY, STRING),
    new Field(CLIENT_ID_KEY, STRING),
    new Field(CLIENT_HOST_KEY, STRING),
    new Field(REBALANCE_TIMEOUT_KEY, INT32),
    new Field(SESSION_TIMEOUT_KEY, INT32),
    new Field(SUBSCRIPTION_KEY, BYTES),
    new Field(ASSIGNMENT_KEY, BYTES))

  private val PROTOCOL_TYPE_KEY = "protocol_type"
  private val GENERATION_KEY = "generation"
  private val PROTOCOL_KEY = "protocol"
  private val LEADER_KEY = "leader"
  private val MEMBERS_KEY = "members"

  private val GROUP_METADATA_VALUE_SCHEMA_V0 = new Schema(
    new Field(PROTOCOL_TYPE_KEY, STRING),
    new Field(GENERATION_KEY, INT32),
    new Field(PROTOCOL_KEY, NULLABLE_STRING),
    new Field(LEADER_KEY, NULLABLE_STRING),
    new Field(MEMBERS_KEY, new ArrayOf(MEMBER_METADATA_V0)))

  private val GROUP_METADATA_VALUE_SCHEMA_V1 = new Schema(
    new Field(PROTOCOL_TYPE_KEY, STRING),
    new Field(GENERATION_KEY, INT32),
    new Field(PROTOCOL_KEY, NULLABLE_STRING),
    new Field(LEADER_KEY, NULLABLE_STRING),
    new Field(MEMBERS_KEY, new ArrayOf(MEMBER_METADATA_V1)))

  // map of versions to key schemas as data types
  private val MESSAGE_TYPE_SCHEMAS = Map(
    0 -> OFFSET_COMMIT_KEY_SCHEMA,
    1 -> OFFSET_COMMIT_KEY_SCHEMA,
    2 -> GROUP_METADATA_KEY_SCHEMA)

  // map of version of offset value schemas
  private val OFFSET_VALUE_SCHEMAS = Map(
    0 -> OFFSET_COMMIT_VALUE_SCHEMA_V0,
    1 -> OFFSET_COMMIT_VALUE_SCHEMA_V1)
  private val CURRENT_OFFSET_VALUE_SCHEMA_VERSION = 1.toShort

  // map of version of group metadata value schemas
  private val GROUP_VALUE_SCHEMAS = Map(
    0 -> GROUP_METADATA_VALUE_SCHEMA_V0,
    1 -> GROUP_METADATA_VALUE_SCHEMA_V1)
  private val CURRENT_GROUP_VALUE_SCHEMA_VERSION = 1.toShort

  private val CURRENT_OFFSET_KEY_SCHEMA = schemaForKey(CURRENT_OFFSET_KEY_SCHEMA_VERSION)
  private val CURRENT_GROUP_KEY_SCHEMA = schemaForKey(CURRENT_GROUP_KEY_SCHEMA_VERSION)

  private val CURRENT_OFFSET_VALUE_SCHEMA = schemaForOffset(CURRENT_OFFSET_VALUE_SCHEMA_VERSION)
  private val CURRENT_GROUP_VALUE_SCHEMA = schemaForGroup(CURRENT_GROUP_VALUE_SCHEMA_VERSION)

  private def schemaForKey(version: Int) = {
    val schemaOpt = MESSAGE_TYPE_SCHEMAS.get(version)
    schemaOpt match {
      case Some(schema) => schema
      case _ => throw new KafkaException("Unknown offset schema version " + version)
    }
  }

  private def schemaForOffset(version: Int) = {
    val schemaOpt = OFFSET_VALUE_SCHEMAS.get(version)
    schemaOpt match {
      case Some(schema) => schema
      case _ => throw new KafkaException("Unknown offset schema version " + version)
    }
  }

  private def schemaForGroup(version: Int) = {
    val schemaOpt = GROUP_VALUE_SCHEMAS.get(version)
    schemaOpt match {
      case Some(schema) => schema
      case _ => throw new KafkaException("Unknown group metadata version " + version)
    }
  }

  /**
   * Generates the key for offset commit message for given (group, topic, partition)
   *
   * @return key for offset commit message
   */
  private[group] def offsetCommitKey(group: String, topicPartition: TopicPartition,
                                           versionId: Short = 0): Array[Byte] = {
    val key = new Struct(CURRENT_OFFSET_KEY_SCHEMA)
    key.set(OFFSET_KEY_GROUP_FIELD, group)
    key.set(OFFSET_KEY_TOPIC_FIELD, topicPartition.topic)
    key.set(OFFSET_KEY_PARTITION_FIELD, topicPartition.partition)

    val byteBuffer = ByteBuffer.allocate(2 /* version */ + key.sizeOf)
    byteBuffer.putShort(CURRENT_OFFSET_KEY_SCHEMA_VERSION)
    key.writeTo(byteBuffer)
    byteBuffer.array()
  }

  /**
   * Generates the key for group metadata message for given group
   *
   * @return key bytes for group metadata message
   */
  private[group] def groupMetadataKey(group: String): Array[Byte] = {
    val key = new Struct(CURRENT_GROUP_KEY_SCHEMA)
    key.set(GROUP_KEY_GROUP_FIELD, group)

    val byteBuffer = ByteBuffer.allocate(2 /* version */ + key.sizeOf)
    byteBuffer.putShort(CURRENT_GROUP_KEY_SCHEMA_VERSION)
    key.writeTo(byteBuffer)
    byteBuffer.array()
  }

  /**
   * Generates the payload for offset commit message from given offset and metadata
   *
   * @param offsetAndMetadata consumer's current offset and metadata
   * @return payload for offset commit message
   */
  private[group] def offsetCommitValue(offsetAndMetadata: OffsetAndMetadata): Array[Byte] = {
    // generate commit value with schema version 1
    val value = new Struct(CURRENT_OFFSET_VALUE_SCHEMA)
    value.set(OFFSET_VALUE_OFFSET_FIELD_V1, offsetAndMetadata.offset)
    value.set(OFFSET_VALUE_METADATA_FIELD_V1, offsetAndMetadata.metadata)
    value.set(OFFSET_VALUE_COMMIT_TIMESTAMP_FIELD_V1, offsetAndMetadata.commitTimestamp)
    value.set(OFFSET_VALUE_EXPIRE_TIMESTAMP_FIELD_V1, offsetAndMetadata.expireTimestamp)
    val byteBuffer = ByteBuffer.allocate(2 /* version */ + value.sizeOf)
    byteBuffer.putShort(CURRENT_OFFSET_VALUE_SCHEMA_VERSION)
    value.writeTo(byteBuffer)
    byteBuffer.array()
  }

  /**
   * Generates the payload for group metadata message from given offset and metadata
   * assuming the generation id, selected protocol, leader and member assignment are all available
   *
   * @param groupMetadata current group metadata
   * @param assignment the assignment for the rebalancing generation
   * @param version the version of the value message to use
   * @return payload for offset commit message
   */
  private[group] def groupMetadataValue(groupMetadata: GroupMetadata,
                                        assignment: Map[String, Array[Byte]],
                                        version: Short = 0): Array[Byte] = {
    val value = if (version == 0) new Struct(GROUP_METADATA_VALUE_SCHEMA_V0) else new Struct(CURRENT_GROUP_VALUE_SCHEMA)

    value.set(PROTOCOL_TYPE_KEY, groupMetadata.protocolType.getOrElse(""))
    value.set(GENERATION_KEY, groupMetadata.generationId)
    value.set(PROTOCOL_KEY, groupMetadata.protocolOrNull)
    value.set(LEADER_KEY, groupMetadata.leaderOrNull)

    val memberArray = groupMetadata.allMemberMetadata.map { memberMetadata =>
      val memberStruct = value.instance(MEMBERS_KEY)
      memberStruct.set(MEMBER_ID_KEY, memberMetadata.memberId)
      memberStruct.set(CLIENT_ID_KEY, memberMetadata.clientId)
      memberStruct.set(CLIENT_HOST_KEY, memberMetadata.clientHost)
      memberStruct.set(SESSION_TIMEOUT_KEY, memberMetadata.sessionTimeoutMs)

      if (version > 0)
        memberStruct.set(REBALANCE_TIMEOUT_KEY, memberMetadata.rebalanceTimeoutMs)

      // The group is non-empty, so the current protocol must be defined
      val protocol = groupMetadata.protocolOrNull
      if (protocol == null)
        throw new IllegalStateException("Attempted to write non-empty group metadata with no defined protocol")

      val metadata = memberMetadata.metadata(protocol)
      memberStruct.set(SUBSCRIPTION_KEY, ByteBuffer.wrap(metadata))

      val memberAssignment = assignment(memberMetadata.memberId)
      assert(memberAssignment != null)

      memberStruct.set(ASSIGNMENT_KEY, ByteBuffer.wrap(memberAssignment))

      memberStruct
    }

    value.set(MEMBERS_KEY, memberArray.toArray)

    val byteBuffer = ByteBuffer.allocate(2 /* version */ + value.sizeOf)
    byteBuffer.putShort(version)
    value.writeTo(byteBuffer)
    byteBuffer.array()
  }

  /**
   * Decodes the offset messages' key
   *
   * @param buffer input byte-buffer
   * @return an GroupTopicPartition object
   */
  def readMessageKey(buffer: ByteBuffer): BaseKey = {
    val version = buffer.getShort
    val keySchema = schemaForKey(version)
    val key = keySchema.read(buffer)

    if (version <= CURRENT_OFFSET_KEY_SCHEMA_VERSION) {
      // version 0 and 1 refer to offset
      val group = key.get(OFFSET_KEY_GROUP_FIELD).asInstanceOf[String]
      val topic = key.get(OFFSET_KEY_TOPIC_FIELD).asInstanceOf[String]
      val partition = key.get(OFFSET_KEY_PARTITION_FIELD).asInstanceOf[Int]

      OffsetKey(version, GroupTopicPartition(group, new TopicPartition(topic, partition)))

    } else if (version == CURRENT_GROUP_KEY_SCHEMA_VERSION) {
      // version 2 refers to offset
      val group = key.get(GROUP_KEY_GROUP_FIELD).asInstanceOf[String]

      GroupMetadataKey(version, group)
    } else {
      throw new IllegalStateException("Unknown version " + version + " for group metadata message")
    }
  }

  /**
   * Decodes the offset messages' payload and retrieves offset and metadata from it
   *
   * @param buffer input byte-buffer
   * @return an offset-metadata object from the message
   */
  def readOffsetMessageValue(buffer: ByteBuffer): OffsetAndMetadata = {
    if (buffer == null) { // tombstone
      null
    } else {
      val version = buffer.getShort
      val valueSchema = schemaForOffset(version)
      val value = valueSchema.read(buffer)

      if (version == 0) {
        val offset = value.get(OFFSET_VALUE_OFFSET_FIELD_V0).asInstanceOf[Long]
        val metadata = value.get(OFFSET_VALUE_METADATA_FIELD_V0).asInstanceOf[String]
        val timestamp = value.get(OFFSET_VALUE_TIMESTAMP_FIELD_V0).asInstanceOf[Long]

        OffsetAndMetadata(offset, metadata, timestamp)
      } else if (version == 1) {
        val offset = value.get(OFFSET_VALUE_OFFSET_FIELD_V1).asInstanceOf[Long]
        val metadata = value.get(OFFSET_VALUE_METADATA_FIELD_V1).asInstanceOf[String]
        val commitTimestamp = value.get(OFFSET_VALUE_COMMIT_TIMESTAMP_FIELD_V1).asInstanceOf[Long]
        val expireTimestamp = value.get(OFFSET_VALUE_EXPIRE_TIMESTAMP_FIELD_V1).asInstanceOf[Long]

        OffsetAndMetadata(offset, metadata, commitTimestamp, expireTimestamp)
      } else {
        throw new IllegalStateException("Unknown offset message version")
      }
    }
  }

  /**
   * Decodes the group metadata messages' payload and retrieves its member metadatafrom it
   *
   * @param buffer input byte-buffer
   * @return a group metadata object from the message
   */
  def readGroupMessageValue(groupId: String, buffer: ByteBuffer): GroupMetadata = {
    if (buffer == null) { // tombstone
      null
    } else {
      val version = buffer.getShort
      val valueSchema = schemaForGroup(version)
      val value = valueSchema.read(buffer)

      if (version == 0 || version == 1) {
        val generationId = value.get(GENERATION_KEY).asInstanceOf[Int]
        val protocolType = value.get(PROTOCOL_TYPE_KEY).asInstanceOf[String]
        val protocol = value.get(PROTOCOL_KEY).asInstanceOf[String]
        val leaderId = value.get(LEADER_KEY).asInstanceOf[String]
        val memberMetadataArray = value.getArray(MEMBERS_KEY)
        val initialState = if (memberMetadataArray.isEmpty) Empty else Stable

        val members = memberMetadataArray.map { memberMetadataObj =>
          val memberMetadata = memberMetadataObj.asInstanceOf[Struct]
          val memberId = memberMetadata.get(MEMBER_ID_KEY).asInstanceOf[String]
          val clientId = memberMetadata.get(CLIENT_ID_KEY).asInstanceOf[String]
          val clientHost = memberMetadata.get(CLIENT_HOST_KEY).asInstanceOf[String]
          val sessionTimeout = memberMetadata.get(SESSION_TIMEOUT_KEY).asInstanceOf[Int]
          val rebalanceTimeout = if (version == 0) sessionTimeout else memberMetadata.get(REBALANCE_TIMEOUT_KEY).asInstanceOf[Int]
          val subscription = Utils.toArray(memberMetadata.get(SUBSCRIPTION_KEY).asInstanceOf[ByteBuffer])

          val member = new MemberMetadata(memberId, groupId, clientId, clientHost, rebalanceTimeout, sessionTimeout,
            protocolType, List((protocol, subscription)))
          member.assignment = Utils.toArray(memberMetadata.get(ASSIGNMENT_KEY).asInstanceOf[ByteBuffer])
          member
        }
        GroupMetadata.loadGroup(groupId, initialState, generationId, protocolType, protocol, leaderId, members)
      } else {
        throw new IllegalStateException("Unknown group metadata message version")
      }
    }
  }

  // Formatter for use with tools such as console consumer: Consumer should also set exclude.internal.topics to false.
  // (specify --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" when consuming __consumer_offsets)
  class OffsetsMessageFormatter extends MessageFormatter {
    def writeTo(consumerRecord: ConsumerRecord[Array[Byte], Array[Byte]], output: PrintStream) {
      Option(consumerRecord.key).map(key => GroupMetadataManager.readMessageKey(ByteBuffer.wrap(key))).foreach {
        // Only print if the message is an offset record.
        // We ignore the timestamp of the message because GroupMetadataMessage has its own timestamp.
        case offsetKey: OffsetKey =>
          val groupTopicPartition = offsetKey.key
          val value = consumerRecord.value
          val formattedValue =
            if (value == null) "NULL"
            else GroupMetadataManager.readOffsetMessageValue(ByteBuffer.wrap(value)).toString
          output.write(groupTopicPartition.toString.getBytes(StandardCharsets.UTF_8))
          output.write("::".getBytes(StandardCharsets.UTF_8))
          output.write(formattedValue.getBytes(StandardCharsets.UTF_8))
          output.write("\n".getBytes(StandardCharsets.UTF_8))
        case _ => // no-op
      }
    }
  }

  // Formatter for use with tools to read group metadata history
  class GroupMetadataMessageFormatter extends MessageFormatter {
    def writeTo(consumerRecord: ConsumerRecord[Array[Byte], Array[Byte]], output: PrintStream) {
      Option(consumerRecord.key).map(key => GroupMetadataManager.readMessageKey(ByteBuffer.wrap(key))).foreach {
        // Only print if the message is a group metadata record.
        // We ignore the timestamp of the message because GroupMetadataMessage has its own timestamp.
        case groupMetadataKey: GroupMetadataKey =>
          val groupId = groupMetadataKey.key
          val value = consumerRecord.value
          val formattedValue =
            if (value == null) "NULL"
            else GroupMetadataManager.readGroupMessageValue(groupId, ByteBuffer.wrap(value)).toString
          output.write(groupId.getBytes(StandardCharsets.UTF_8))
          output.write("::".getBytes(StandardCharsets.UTF_8))
          output.write(formattedValue.getBytes(StandardCharsets.UTF_8))
          output.write("\n".getBytes(StandardCharsets.UTF_8))
        case _ => // no-op
      }
    }
  }

}

case class GroupTopicPartition(group: String, topicPartition: TopicPartition) {

  def this(group: String, topic: String, partition: Int) =
    this(group, new TopicPartition(topic, partition))

  override def toString: String =
    "[%s,%s,%d]".format(group, topicPartition.topic, topicPartition.partition)
}

trait BaseKey{
  def version: Short
  def key: Any
}

case class OffsetKey(version: Short, key: GroupTopicPartition) extends BaseKey {

  override def toString: String = key.toString
}

case class GroupMetadataKey(version: Short, key: String) extends BaseKey {

  override def toString: String = key
}


text:
from .db import db
from flask_security import UserMixin, RoleMixin
from flask_security.utils import encrypt_password

class Role(db.Document, RoleMixin):

    name = db.StringField(max_length=80, unique=True)
    description = db.StringField(max_length=255)

    @classmethod
    def createrole(cls, name, description=None):
        return cls.objects.create(
            name=name,
            description=description
        )

class User(db.Document, UserMixin):
    name = db.StringField(max_length=255)
    email = db.EmailField(max_length=255, unique=True)
    password = db.StringField(max_length=255)
    active = db.BooleanField(default=True)
    confirmed_at = db.DateTimeField()
    roles = db.ListField(
        db.ReferenceField(Role, reverse_delete_rule=db.DENY), default=[]
    )
    last_login_at = db.DateTimeField()
    current_login_at = db.DateTimeField()
    last_login_ip = db.StringField(max_length=255)
    current_login_ip = db.StringField(max_length=255)
    login_count = db.IntField()

    @classmethod
    def createuser(cls, name, email, password,
                   active=True, roles=None, username=None,
                   *args, **kwargs):
        return cls.objects.create(
            name=name,
            email=email,
            password=encrypt_password(password),
            active=active,
            roles=roles,
            username=username,
            *args,
            **kwargs
        )


text:
<AttributeMetadata xmlns:i="http://www.w3.org/2001/XMLSchema-instance" i:type="MoneyAttributeMetadata" xmlns="http://schemas.microsoft.com/xrm/2011/Metadata">
	<AttributeOf i:nil="true" />
	<AttributeType>Money</AttributeType>
	<Description xmlns:d2p1="http://schemas.microsoft.com/xrm/2011/Contracts">
		<d2p1:LocalizedLabels>
			<d2p1:LocalizedLabel>
				<d2p1:Label>Type the credit limit of the contact for reference when you address invoice and accounting issues with the customer.</d2p1:Label>
				<d2p1:LanguageCode>1033</d2p1:LanguageCode>
			</d2p1:LocalizedLabel>
		</d2p1:LocalizedLabels>
	</Description>
	<DisplayName xmlns:d2p1="http://schemas.microsoft.com/xrm/2011/Contracts">
		<d2p1:LocalizedLabels>
			<d2p1:LocalizedLabel>
				<d2p1:Label>Credit Limit</d2p1:Label>
				<d2p1:LanguageCode>1033</d2p1:LanguageCode>
			</d2p1:LocalizedLabel>
		</d2p1:LocalizedLabels>
	</DisplayName>
	<EntityLogicalName>contact</EntityLogicalName>
	<InheritsFrom i:nil="true" />
	<IsCustomAttribute>false</IsCustomAttribute>
	<IsFilterable>false</IsFilterable>
	<IsPrimaryId>false</IsPrimaryId>
	<IsPrimaryName>false</IsPrimaryName>
	<LinkedAttributeId i:nil="true" />
	<LogicalName>creditlimit</LogicalName>
	<RequiredLevel>
		<CanBeChanged xmlns="http://schemas.microsoft.com/xrm/2011/Contracts">true</CanBeChanged>
		<ManagedPropertyLogicalName xmlns="http://schemas.microsoft.com/xrm/2011/Contracts">canmodifyrequirementlevelsettings</ManagedPropertyLogicalName>
		<Value xmlns="http://schemas.microsoft.com/xrm/2011/Contracts">None</Value>
	</RequiredLevel>
	<SchemaName>CreditLimit</SchemaName>
	<AttributeTypeName xmlns:d2p1="http://schemas.microsoft.com/xrm/2013/Metadata">
		<d2p1:Value>MoneyType</d2p1:Value>
	</AttributeTypeName>
	<IsLogical>false</IsLogical>
	<SourceType>0</SourceType>
	<CalculationOf i:nil="true" />
	<ImeMode>Disabled</ImeMode>
	<MaxValue>100000000000000</MaxValue>
	<MinValue>0</MinValue>
	<Precision>2</Precision>
	<PrecisionSource>2</PrecisionSource>
	<FormulaDefinition i:nil="true" />
	<IsBaseCurrency>false</IsBaseCurrency>
	<SourceTypeMask>0</SourceTypeMask>
</AttributeMetadata>


text:
.class public interface abstract Lcom/android/systemui/statusbar/phone/taskbar/desk/DeskLaunchableIconView$Mouse;
.super Ljava/lang/Object;
.source "DeskLaunchableIconView.java"

# annotations
.annotation system Ldalvik/annotation/EnclosingClass;
    value = Lcom/android/systemui/statusbar/phone/taskbar/desk/DeskLaunchableIconView;
.end annotation

.annotation system Ldalvik/annotation/InnerClass;
    accessFlags = 0x609
    name = "Mouse"
.end annotation

# virtual methods
.method public abstract onLeftClick(Lcom/android/systemui/statusbar/phone/taskbar/desk/DeskLaunchableIconView;)V
.end method

.method public abstract onRightClick(Lcom/android/systemui/statusbar/phone/taskbar/desk/DeskLaunchableIconView;)V
.end method

.method public abstract shouldShowTooltip(Lcom/android/systemui/statusbar/phone/taskbar/desk/DeskLaunchableIconView;)Z
.end method


text:
import { Injectable, TemplateRef } from '@angular/core';

/**
 * A service to open modals
 */
@Injectable()
export class ModalService {

  /**
   * A method to open a modal
   */
  open(content: string | TemplateRef<any>, options = {}): Promise<any> {
    return Promise.resolve();
  }

  /**
   * Checks if a modal is open
   */
  isOpen(): boolean {
    return false;
  }
}


text:
import { createSelector } from 'reselect';
/**
 * Direct selector to the available state domain
 */
const selectAvailableDomain = () => (state) => state.get('available');

const selectAvailableState = () => (state) => state.get('available').toJS();

/**
 * Other specific selectors
 */
 // const makeSelectCarstoAvailable = ()=> createSelector(
 //   selectHomePageDomain(),
 //   (state) => state.get('cars')
 // )
/**
 * Default selector used by Available
 */

const makeSelectAvailable = () => createSelector(
  selectAvailableDomain(),
  (substate) => substate.toJS()
);

export default makeSelectAvailable;
export {
  selectAvailableDomain,
  selectAvailableState,
  // makeSelectCarstoAvailable
};


text:
namespace media {

FakeVideoCaptureDeviceFactory::FakeVideoCaptureDeviceFactory()
    : number_of_devices_(1) {
}

scoped_ptr<VideoCaptureDevice> FakeVideoCaptureDeviceFactory::Create(
    const VideoCaptureDevice::Name& device_name) {
  DCHECK(thread_checker_.CalledOnValidThread());

  const std::string option =
      base::CommandLine::ForCurrentProcess()->GetSwitchValueASCII(
          switches::kUseFakeDeviceForMediaStream);

  const FakeVideoCaptureDevice::BufferOwnership fake_vcd_ownership =
      base::StartsWith(option, "client", base::CompareCase::INSENSITIVE_ASCII)
          ? FakeVideoCaptureDevice::BufferOwnership::CLIENT_BUFFERS
          : FakeVideoCaptureDevice::BufferOwnership::OWN_BUFFERS;

  const FakeVideoCaptureDevice::BufferPlanarity fake_vcd_planarity =
      base::EndsWith(option, "triplanar", base::CompareCase::INSENSITIVE_ASCII)
          ? FakeVideoCaptureDevice::BufferPlanarity::TRIPLANAR
          : FakeVideoCaptureDevice::BufferPlanarity::PACKED;

  for (int n = 0; n < number_of_devices_; ++n) {
    std::string possible_id = base::StringPrintf("/dev/video%d", n);
    if (device_name.id().compare(possible_id) == 0) {
      return scoped_ptr<VideoCaptureDevice>(
          new FakeVideoCaptureDevice(fake_vcd_ownership, fake_vcd_planarity));
    }
  }
  return scoped_ptr<VideoCaptureDevice>();
}

void FakeVideoCaptureDeviceFactory::GetDeviceNames(
    VideoCaptureDevice::Names* const device_names) {
  DCHECK(thread_checker_.CalledOnValidThread());
  DCHECK(device_names->empty());
  for (int n = 0; n < number_of_devices_; ++n) {
    VideoCaptureDevice::Name name(base::StringPrintf("fake_device_%d", n),
                                  base::StringPrintf("/dev/video%d", n)
#if defined(OS_LINUX)
                                      ,
                                  VideoCaptureDevice::Name::V4L2_SINGLE_PLANE
#elif defined(OS_MACOSX)
                                      ,
                                  VideoCaptureDevice::Name::AVFOUNDATION
#elif defined(OS_WIN)
                                      ,
                                  VideoCaptureDevice::Name::DIRECT_SHOW
#elif defined(OS_ANDROID)
                                      ,
                                  VideoCaptureDevice::Name::API2_LEGACY
#endif
                                  );
    device_names->push_back(name);
  }
}

void FakeVideoCaptureDeviceFactory::GetDeviceSupportedFormats(
    const VideoCaptureDevice::Name& device,
    VideoCaptureFormats* supported_formats) {
  DCHECK(thread_checker_.CalledOnValidThread());
  const int frame_rate = 1000 / FakeVideoCaptureDevice::FakeCapturePeriodMs();
  const gfx::Size supported_sizes[] = {gfx::Size(320, 240),
                                       gfx::Size(640, 480),
                                       gfx::Size(1280, 720),
                                       gfx::Size(1920, 1080)};
  supported_formats->clear();
  for (const auto& size : supported_sizes) {
    supported_formats->push_back(VideoCaptureFormat(
        size, frame_rate, media::VIDEO_CAPTURE_PIXEL_FORMAT_I420));
  }
}

}  // namespace media


text:
<?php

namespace DCarbone\PHPFHIRGenerated\R4\PHPFHIRTests\FHIRElement\FHIRBackboneElement\FHIRConsent;

use PHPUnit\Framework\TestCase;
use DCarbone\PHPFHIRGenerated\R4\FHIRElement\FHIRBackboneElement\FHIRConsent\FHIRConsentVerification;

/**
 * Class FHIRConsentVerificationTest
 * @package \DCarbone\PHPFHIRGenerated\R4\PHPFHIRTests\FHIRElement\FHIRBackboneElement\FHIRConsent
 */
class FHIRConsentVerificationTest extends TestCase
{
    public function testCanConstructTypeNoArgs()
    {
        $type = new FHIRConsentVerification();
        $this->assertInstanceOf('\DCarbone\PHPFHIRGenerated\R4\FHIRElement\FHIRBackboneElement\FHIRConsent\FHIRConsentVerification', $type);
    }
}


text:
use std::ops::Deref;
use glium::Display;

use layout::LayoutBuffer;
use resource::ResourceManager;
use util::BufferFromTree;
use state::StateBuffer;
use state::StateData;

use super::RenderData;
use super::TextureRule;

pub struct RenderBuffer {
    render_data: BufferFromTree<RenderData>,
}

impl Deref for RenderBuffer {
    type Target = [RenderData];

    fn deref<'a>(&'a self) -> &'a [RenderData] {
        self.render_data.deref()
    }
}

impl RenderBuffer {

    pub fn new<R>(
        display: &Display,
        resource_manager: &R,
        state_buffer: &StateBuffer) -> RenderBuffer
        where R: ResourceManager
    {
        let node_producer = |state: &StateData| {
            if let Some(img) = state.get_background_image() {
                let rule = state.get_background_rule().unwrap_or(TextureRule::Fit);
                Some(RenderData::new(
                    display,
                    resource_manager,
                    img,
                    rule
                ))
            } else {
                None
            }
        };

        RenderBuffer {
            render_data: BufferFromTree::from_buffer(
                state_buffer,
                node_producer
            )
        }
    }

    pub fn update_from_state<R: ResourceManager>(
        &mut self,
        display: &Display,
        resource_manager: &R,
        state_data: &StateBuffer)
    {

        for (&i, data) in self.render_data.enumerate_lookup_indices_mut().unwrap() {

            let state = unsafe { state_data.get_unchecked(i) };

            if let Some(img) = state.get_background_image() {

                data.update_texture(display, resource_manager, img);
            }
        }
    }

    pub fn update_from_layout(&mut self, display: &Display, layout_data: &LayoutBuffer) {

        for (&i, data) in self.render_data.enumerate_lookup_indices_mut().unwrap() {
            // This part is always safe because the initialization step
            // ensure that:
            //       self.layout_data.len() >= self.render_data
            let boxi = unsafe { layout_data.get_unchecked(i) };
            data.update_coords(display, &boxi);
        }
    }
}

// ======================================== //
//                   TESTS                  //
// ======================================== //

#[cfg(test)]
mod test {

    use super::RenderBuffer;
    use std::io::BufReader;
    use std::path::PathBuf;
    use glutin::WindowBuilder;
    use glium::DisplayBuild;
    use markup::{self, Node};
    use style;
    use state::StateBuffer;
    use oil_shared::style::Stylesheet;
    use oil_shared::deps::{Constructor, StyleDefinitions};
    use oil_parsers::{StdOutErrorReporter};
    use resource::{self, ResourceManager};

    fn stylesheet<R: ResourceManager>(st: &str, r: &mut R) -> Stylesheet {
        let reader = BufReader::new(st.as_bytes());
        let mut defs = StyleDefinitions::new();
        defs.insert("toto".to_string(),
            Constructor::Image(PathBuf::new(), None, None, None, None));
        style::parse(StdOutErrorReporter, reader, &defs, r)
    }

    fn markup_tree(mk: &str) -> markup::Node {
        let reader = BufReader::new(mk.as_bytes());
        let lib = markup::parse(StdOutErrorReporter, reader);
        let (_, root) = lib.views.into_iter().next().unwrap();
        root
    }

    // TODO(Nemikolh): Fix that test by allowing glium in headless.
    //                 Will require some refactoring.
    //#[test]
    // fn lookup_table_should_contain_correct_indices() {
    //
    //     let mut fake_resource_manager = resource::create_null_manager();
    //     let stylesheet = stylesheet(
    //         ".btn { background-image: $toto; }",
    //         &mut fake_resource_manager);
    //     let root = markup_tree(
    //         "<view>\
    //             <button class=\"btn\"></button>\
    //             <button class=\"\"></button>\
    //             <button class=\"btn\"></button>\
    //         </view>
    //         ");
    //     let state_buffer = StateBuffer::new(&root, &stylesheet);
    //
    //     // FIXME(This match is here to have the travis test pass)
    //     if let Some(fake_display) = WindowBuilder::new()
    //         .with_dimensions(1, 1).with_visibility(false).build_glium().ok()
    //     {
    //
    //         let mut buffer = RenderBuffer::new(
    //             &fake_display,
    //             &fake_resource_manager,
    //             &state_buffer
    //         );
    //
    //         assert_eq!(buffer.render_data.len(), 2);
    //         let mut iter = buffer.render_data.enumerate_lookup_indices_mut().unwrap();
    //         let (&i, _) = iter.next().unwrap();
    //         assert_eq!(i, 1);
    //         let (&j, _) = iter.next().unwrap();
    //         assert_eq!(j, 3);
    //     }
    // }
}


text:
package com.querydsl.jpa;

import java.util.Map;

import javax.annotation.Nullable;

import com.google.common.collect.Maps;
import com.querydsl.core.JoinType;
import com.querydsl.core.QueryMetadata;
import com.querydsl.core.support.ReplaceVisitor;
import com.querydsl.core.types.*;
import com.querydsl.core.types.dsl.Expressions;

class JPAMapAccessVisitor extends ReplaceVisitor<Void> {

    private final QueryMetadata metadata;

    private final Map<Expression<?>, Path<?>> aliases;

    private final Map<Path<?>, Path<?>> replacements = Maps.newHashMap();

    public JPAMapAccessVisitor(QueryMetadata metadata, Map<Expression<?>, Path<?>> aliases) {
        this.metadata = metadata;
        this.aliases = aliases;
    }

    @SuppressWarnings("unchecked")
    @Override
    public Expression<?> visit(Operation<?> expr, @Nullable Void context) {
        if (expr.getOperator() == Ops.CONTAINS_KEY) {
            ParameterizedExpression map = (ParameterizedExpression<?>) expr.getArg(0);
            Expression key = expr.getArg(1);
            Path replacement = ExpressionUtils.path(map.getParameter(1),
                    ExpressionUtils.createRootVariable((Path<?>) map, Math.abs(expr.hashCode())));
            metadata.addJoin(JoinType.LEFTJOIN, ExpressionUtils.as(map, replacement));
            metadata.addJoinCondition(ExpressionUtils.eq(
                    Expressions.operation(map.getParameter(0), JPQLOps.KEY, replacement),
                    key));
            return ExpressionUtils.isNotNull(replacement);
        } else if (expr.getOperator() == Ops.CONTAINS_VALUE) {
            ParameterizedExpression<?> map = (ParameterizedExpression<?>) expr.getArg(0);
            Expression<?> value = expr.getArg(1);
            return Expressions.predicate(JPQLOps.MEMBER_OF, value, map);
        } else {
            return super.visit(expr, context);
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public Expression<?> visit(Path<?> expr, @Nullable Void context) {
        expr = (Path<?>) super.visit(expr, null);
        PathMetadata pathMetadata = expr.getMetadata();
        if (pathMetadata.getPathType() == PathType.MAPVALUE
         || pathMetadata.getPathType() == PathType.MAPVALUE_CONSTANT) {
            Path<?> replacement = replacements.get(expr);
            if (replacement == null) {
                // join parent as path123 on key(path123) = ...
                Path parent = shorten(pathMetadata.getParent(), true);
                ParameterizedExpression parExpr = (ParameterizedExpression) pathMetadata.getParent();
                replacement = ExpressionUtils.path(parExpr.getParameter(1),
                        ExpressionUtils.createRootVariable(parent, replacements.size()));
                metadata.addJoin(JoinType.LEFTJOIN, ExpressionUtils.as(parent, replacement));
                metadata.addJoinCondition(ExpressionUtils.eq(
                        Expressions.operation(parExpr.getParameter(0), JPQLOps.KEY, replacement),
                        ExpressionUtils.toExpression(pathMetadata.getElement())));
                replacements.put(expr, replacement);
            }
            return replacement;
        } else {
            return super.visit(expr, context);
        }
    }

    /**
     * Shorten the parent path to a length of max 2 elements
     */
    private Path<?> shorten(Path<?> path, boolean outer) {
        if (aliases.containsKey(path)) {
            return aliases.get(path);
        } else if (path.getMetadata().isRoot()) {
            return path;
        } else if (path.getMetadata().getParent().getMetadata().isRoot() && outer) {
            return path;
        } else {
            Class<?> type = JPAQueryMixin.getElementTypeOrType(path);
            Path<?> parent = shorten(path.getMetadata().getParent(), false);
            Path oldPath = ExpressionUtils.path(path.getType(),
                    new PathMetadata(parent, path.getMetadata().getElement(), path.getMetadata().getPathType()));
            if (oldPath.getMetadata().getParent().getMetadata().isRoot() && outer) {
                return oldPath;
            } else {
                Path newPath = ExpressionUtils.path(type, ExpressionUtils.createRootVariable(oldPath));
                aliases.put(path, newPath);
                metadata.addJoin(JoinType.LEFTJOIN, ExpressionUtils.as(oldPath, newPath));
                return newPath;
            }
        }
    }

}


text:
<?php

declare(strict_types=1);

namespace Phpfastcache\Drivers\Memstatic;

use Phpfastcache\Config\ConfigurationOption;

class Config extends ConfigurationOption
{

}


text:
import json
import logging
import os
import requests
import subprocess
import urllib.parse

_logger = logging.getLogger('fireci.uploader')

def post_report(test_report, metrics_service_url, access_token, metric_type):
  """Post a report to the metrics service backend."""

  endpoint = ''
  if os.getenv('GITHUB_ACTIONS'):
    endpoint = _construct_request_endpoint_for_github_actions(metric_type)
  elif os.getenv('PROW_JOB_ID'):
    endpoint = _construct_request_endpoint_for_prow(metric_type)

  headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'}
  data = json.dumps(test_report)

  _logger.info('Posting to the metrics service ...')
  _logger.info(f'Request endpoint: {endpoint}')
  _logger.info(f'Request data: {data}')

  request_url = f'{metrics_service_url}{endpoint}'
  result = requests.post(request_url, data=data, headers=headers)

  _logger.info(f'Response: {result.text}')

def _construct_request_endpoint_for_github_actions(metric_type):
  repo = os.getenv('GITHUB_REPOSITORY')
  commit = os.getenv('GITHUB_SHA')
  event_name = os.getenv('GITHUB_EVENT_NAME')

  endpoint = f'/repos/{repo}/commits/{commit}/{metric_type}'
  if event_name == 'pull_request':
    pull_request = os.getenv('GITHUB_PULL_REQUEST_NUMBER')
    base_commit = _get_commit_hash('HEAD^1')
    head_commit = _get_commit_hash('HEAD^2')
    endpoint += f'?pull_request={pull_request}&base_commit={base_commit}&head_commit={head_commit}'
  else:
    branch = os.getenv('GITHUB_REF_NAME')
    endpoint += f'?branch={branch}'

  return endpoint

def _construct_request_endpoint_for_prow(metric_type):
  repo_owner = os.getenv('REPO_OWNER')
  repo_name = os.getenv('REPO_NAME')
  commit = _get_commit_hash('HEAD@{0}')
  pull_request = os.getenv('PULL_NUMBER')

  endpoint = f'/repos/{repo_owner}/{repo_name}/commits/{commit}/{metric_type}'
  if pull_request:
    base_commit = os.getenv('PULL_BASE_SHA')
    head_commit = os.getenv('PULL_PULL_SHA')
    endpoint += f'&pull_request={pull_request}&base_commit={base_commit}&head_commit={head_commit}'
  else:
    branch = os.getenv('PULL_BASE_REF')
    endpoint += f'&branch={branch}'

  return endpoint

def _get_commit_hash(revision):
  result = subprocess.run(['git', 'rev-parse', revision], stdout=subprocess.PIPE, check=True)
  return result.stdout.decode('utf-8').strip()


text:
import os
from os.path import (
    dirname,
    join,
    isfile,
)
import unittest

from webtest import TestApp
from mock import patch

from sqlalchemy import engine_from_config
from sqlalchemy.orm.session import Session
from pyramid.security import has_permission
from pyramid import (
    paster,
    testing,
)

import intranet3

settings_file = os.environ.get('TEST_SETTINGS_PATH')

# creating memcache before other modules are loaded :(
if settings_file and isfile(settings_file):
    settings = paster.get_appsettings(settings_file)
    intranet3.init_memcache(settings)
    intranet3.init_config(settings)
else:
    settings = None

from intranet3 import models as intranet_models
from intranet3.testing import mocks
from intranet3.testing.factory import FactoryMixin

connection = None
engine = None
app = None

def setup_module():
    global connection, app, engine
    if settings is None:
        raise Exception('Settings file not found %s' % settings_file)

    engine = engine_from_config(settings, prefix='sqlalchemy.')

    connection = engine.connect()
    intranet_models.Base.metadata.drop_all(engine)

    intranet_models.DBSession.configure(bind=engine)
    intranet_models.Base.metadata.create_all(engine)

    app = TestApp(paster.get_app(settings_file))

def teardown_module():
    global connection, engine

    intranet_models.Base.metadata.drop_all(engine)
    connection.close()

class IntranetBaseTest(unittest.TestCase):

    def setUp(self):
        self._transaction = connection.begin()
        intranet_models.DBSession_ = Session(connection)
        intranet3.memcache.clear()

    def tearDown(self):
        intranet_models.DBSession_.close()
        self._transaction.rollback()

class IntranetTest(IntranetBaseTest):

    def setUp(self):
        super(IntranetTest, self).setUp()
        self.config = testing.setUp()

        self.request = Request()
        self.request.user = intranet_models.User()
        ac = intranet_models.ApplicationConfig(
            office_ip='192.168.1.1',
            google_user_email='google_user_email@example.com',
            google_user_password='google_user_password',
            holidays_spreadsheet='a',
            hours_employee_project=102,
        )
        intranet_models.DBSession.add(ac)
        intranet_models.DBSession.flush()

    def tearDown(self):
        super(IntranetTest, self).tearDown()
        testing.tearDown()

class IntranetWebTest(IntranetBaseTest):

    def tearDown(self):
        super(IntranetWebTest, self).tearDown()
        app.reset()

    def get(self, *args, **kwargs):
        return app.get(*args, **kwargs)

    def post(self, *args, **kwargs):
        return app.post(*args, **kwargs)

    def post_json(self, *args, **kwargs):
        return app.post_json(*args, **kwargs)

    def put(self, *args, **kwargs):
        return app.put(*args, **kwargs)

    def put_json(self, *args, **kwargs):
        return app.put_json(*args, **kwargs)

    def delete(self, *args, **kwargs):
        return app.delete(*args, **kwargs)

    def delete_json(self, *args, **kwargs):
        return app.delete_json(*args, **kwargs)

    def login(self, name, email):
        with patch("intranet3.views.auth.requests.get") as request_get:
            request_get.return_value = mocks.request_get(name, email)
            with patch(
                "intranet3.views.auth.OAuth2WebServerFlow.step2_exchange"
            ) as flow:
                flow.return_value = mocks.MockOAuth2FlowSampleData()
                with patch("intranet3.views.auth.ApplicationConfig") as config:
                    config.return_value = mocks.MockApplicationConfig()
                    app.get('/auth/callback')

class Request(testing.DummyRequest):

    def __init__(self):
        self.tmpl_ctx = {}

        super(Request, self).__init__()

    def has_perm(self, perm):
        return has_permission(perm, self.context, self)


text:
FOUNDATION_EXPORT double Pods_TopTabBarViewExampleVersionNumber;
FOUNDATION_EXPORT const unsigned char Pods_TopTabBarViewExampleVersionString[];


text: 1. Route protection


text:
// Copyright 2013 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

using VsChromium.Core.Ipc;

namespace VsChromium.Server.Ipc {
  public interface IIpcRequestDispatcher {
    void ProcessRequestAsync(IpcRequest request);
  }
}


text:
// +build !appengine

package buffer

import (
	"fmt"
	"sync"
	"testing"
	"time"
)

func TestCircularBufferSerial(t *testing.T) {
	var size, i uint32
	var result []interface{}

	size = 1 << 15
	cb, err := NewCircularBuffer(size)
	if err != nil {
		t.Fatalf("error allocating CircularBuffer: %v", err)
	}

	for i = 0; i < size/2; i++ {
		cb.Push(i)
	}

	result = cb.Drain()
	if uint32(len(result)) != size/2 {
		t.Fatalf("len(result) = %d; want %d", len(result), size/2)
	}

	// The returned result isn't necessarily sorted.
	seen := make(map[uint32]bool)
	for _, r := range result {
		seen[r.(uint32)] = true
	}

	for i = 0; i < uint32(len(result)); i++ {
		if !seen[i] {
			t.Fatalf("seen[%d] = false; want true", i)
		}
	}

	for i = 0; i < size; i++ {
		cb.Push(i)
	}

	result = cb.Drain()
	if uint32(len(result)) != size {
		t.Fatalf("len(result) = %d; want %d", len(result), size/2)
	}
}

func TestCircularBufferOverflow(t *testing.T) {
	var size, i uint32
	var result []interface{}

	size = 1 << 10
	cb, err := NewCircularBuffer(size)
	if err != nil {
		t.Fatalf("error allocating CircularBuffer: %v", err)
	}

	for i = 0; i < 10*size; i++ {
		cb.Push(i)
	}

	result = cb.Drain()

	if uint32(len(result)) != size {
		t.Fatalf("len(result) = %d; want %d", len(result), size)
	}

	for idx, x := range result {
		if x.(uint32) < size {
			t.Fatalf("result[%d] = %d; want it to be >= %d", idx, x, size)
		}
	}
}

func TestCircularBufferConcurrent(t *testing.T) {
	for tn := 0; tn < 2; tn++ {
		var size uint32
		var result []interface{}

		size = 1 << 6
		cb, err := NewCircularBuffer(size)
		if err != nil {
			t.Fatalf("error allocating CircularBuffer: %v", err)
		}

		type item struct {
			R uint32
			N uint32
			T time.Time
		}

		var wg sync.WaitGroup
		for r := uint32(0); r < 1024; r++ {
			wg.Add(1)
			go func(r uint32) {
				for n := uint32(0); n < size; n++ {
					cb.Push(item{R: r, N: n, T: time.Now()})
				}
				wg.Done()
			}(r)
		}

		// Wait for all goroutines to finish only in one test. Draining
		// concurrently while Pushes are still happening will test for races in the
		// Draining lock.
		if tn == 0 {
			wg.Wait()
		}

		result = cb.Drain()

		// Can't expect the buffer to be full if the Pushes aren't necessarily done.
		if tn == 0 {
			if uint32(len(result)) != size {
				t.Fatalf("len(result) = %d; want %d", len(result), size)
			}
		}

		// There can be absolutely no expectation on the order of the data returned
		// by Drain because: (a) everything is happening concurrently (b) a
		// round-robin is used to write to different queues (and therefore
		// different cachelines) for less write contention.

		// Wait for all goroutines to complete before moving on to other tests. If
		// the benchmarks run after this, it might affect performance unfairly.
		wg.Wait()
	}
}

func BenchmarkCircularBuffer(b *testing.B) {
	x := 1
	for size := 1 << 16; size <= 1<<20; size <<= 1 {
		for routines := 1; routines <= 1<<8; routines <<= 1 {
			b.Run(fmt.Sprintf("goroutines:%d/size:%d", routines, size), func(b *testing.B) {
				cb, err := NewCircularBuffer(uint32(size))
				if err != nil {
					b.Fatalf("error allocating CircularBuffer: %v", err)
				}

				perRoutine := b.N / routines
				var wg sync.WaitGroup
				for r := 0; r < routines; r++ {
					wg.Add(1)
					go func() {
						for i := 0; i < perRoutine; i++ {
							cb.Push(&x)
						}
						wg.Done()
					}()
				}
				wg.Wait()
			})
		}
	}
}


text:
Ext.define('KitchenSink.view.binding.ComponentState', {
    extend: 'Ext.panel.Panel',
    xtype: 'binding-component-state',
    //<example>
    exampleDescription: [
        '
This example shows how to use basic data binding to bind the state of one component to the state of another.
'
    ],
    //</example>

    width: 300,
    layout: 'anchor',
    viewModel: true,
    title: 'Sign Up Form',
    bodyPadding: 10,

    items: [{
        xtype: 'checkbox',
        boxLabel: 'Is Admin',
        reference: 'isAdmin'
    },{
        xtype: 'textfield',
        fieldLabel: 'Admin Key',
        anchor: '0',
        bind: {
            disabled: '{!isAdmin.checked}'
        }
    }]
});