text:
<?php

namespace Oro\Bundle\CalendarBundle\Controller;

use Symfony\Bundle\FrameworkBundle\Controller\Controller;

use Sensio\Bundle\FrameworkExtraBundle\Configuration\Route;
use Sensio\Bundle\FrameworkExtraBundle\Configuration\Template;

use Oro\Bundle\SecurityBundle\Annotation\Acl;
use Oro\Bundle\SecurityBundle\Annotation\AclAncestor;
use Oro\Bundle\SecurityBundle\SecurityFacade;

use Oro\Bundle\CalendarBundle\Entity\Calendar;
use Oro\Bundle\CalendarBundle\Entity\Repository\CalendarRepository;
use Oro\Bundle\CalendarBundle\Provider\CalendarDateTimeConfigProvider;

use Oro\Bundle\UserBundle\Entity\User;
use Oro\Bundle\OrganizationBundle\Entity\Organization;

class CalendarController extends Controller
{
    /**
     * View user's default calendar
     *
     * @Route("/default", name="oro_calendar_view_default")
     * @Template
     * @AclAncestor("oro_calendar_view")
     */
    public function viewDefaultAction()
    {
        /** @var User $user */
        $user = $this->getUser();

        /** @var Organization $organization */
        $organization = $this->get('oro_security.security_facade')->getOrganization();

        $em = $this->getDoctrine()->getManager();
        /** @var CalendarRepository $repo */
        $repo     = $em->getRepository('OroCalendarBundle:Calendar');

        $calendar = $repo->findDefaultCalendar($user->getId(), $organization->getId());

        return $this->viewAction($calendar);
    }

    /**
     * View calendar
     *
     * @Route("/view/{id}", name="oro_calendar_view", requirements={"id"="\d+"})
     *
     * @Template
     * @Acl(
     *      id="oro_calendar_view",
     *      type="entity",
     *      class="OroCalendarBundle:Calendar",
     *      permission="VIEW",
     *      group_name=""
     * )
     */
    public function viewAction(Calendar $calendar)
    {
        /** @var SecurityFacade $securityFacade */
        $securityFacade = $this->get('oro_security.security_facade');
        /** @var CalendarDateTimeConfigProvider $calendarConfigProvider */
        $calendarConfigProvider = $this->get('oro_calendar.provider.calendar_config');

        $dateRange = $calendarConfigProvider->getDateRange();

        $result = array(
            'event_form' => $this->get('oro_calendar.calendar_event.form.template')->createView(),
            'user_select_form' => $this->get('form.factory')
                ->createNamed(
                    'new_calendar',
                    'oro_user_select',
                    null,
                    array(
                        'autocomplete_alias' => 'user_calendars',

                        'configs' => array(
                            'entity_id'               => $calendar->getId(),
                            'entity_name'             => 'OroCalendarBundle:Calendar',
                            'excludeCurrent'          => true,
                            'component'               => 'acl-user-autocomplete',
                            'permission'              => 'VIEW',
                            'placeholder'             => 'oro.calendar.form.choose_user_to_add_calendar',
                            'result_template_twig'    => 'OroCalendarBundle:Calendar:Autocomplete/result.html.twig',
                            'selection_template_twig' => 'OroCalendarBundle:Calendar:Autocomplete/selection.html.twig',
                        ),

                        'grid_name' => 'users-calendar-select-grid-exclude-owner',
                        'random_id' => false,
                        'required'  => true,
                    )
                )
                ->createView(),
            'entity' => $calendar,
            'calendar' => array(
                'selectable' => $securityFacade->isGranted('oro_calendar_event_create'),
                'editable' => $securityFacade->isGranted('oro_calendar_event_update'),
                'removable' => $securityFacade->isGranted('oro_calendar_event_delete'),
                'timezoneOffset' => $calendarConfigProvider->getTimezoneOffset()
            ),
            'startDate' => $dateRange['startDate'],
            'endDate' => $dateRange['endDate'],
        );

        return $result;
    }
}


text:
#ifndef __OPENSPACE_MODULE_GLOBEBROWSING___SHADOWCOMPONENT___H__
#define __OPENSPACE_MODULE_GLOBEBROWSING___SHADOWCOMPONENT___H__

#include <openspace/properties/propertyowner.h>

#include <openspace/camera/camera.h>
#include <openspace/properties/stringproperty.h>
#include <openspace/properties/triggerproperty.h>
#include <openspace/properties/scalar/boolproperty.h>
#include <openspace/properties/scalar/floatproperty.h>
#include <openspace/properties/scalar/intproperty.h>
#include <openspace/properties/vector/vec2property.h>
#include <openspace/properties/vector/vec4property.h>
#include <ghoul/glm.h>
#include <ghoul/opengl/texture.h>
#include <ghoul/opengl/uniformcache.h>
#include <string>
#include <sstream>

namespace ghoul { class Dictionary; }
namespace ghoul::filesystem { class File; }
namespace ghoul::opengl { class ProgramObject; }

namespace openspace {
    struct RenderData;
    struct UpdateData;

namespace documentation { struct Documentation; }

class ShadowComponent : public properties::PropertyOwner {
public:
    struct ShadowMapData {
        glm::dmat4 shadowMatrix = glm::dmat4(1.0);
        GLuint shadowDepthTexture = 0;
    };

    ShadowComponent(const ghoul::Dictionary& dictionary);

    void initialize();
    void initializeGL();
    void deinitializeGL();
    //bool deinitialize();

    bool isReady() const;

    RenderData begin(const RenderData& data);
    void end();
    void update(const UpdateData& data);

    static documentation::Documentation Documentation();

    bool isEnabled() const;

    ShadowComponent::ShadowMapData shadowMapData() const;

    GLuint dDepthTexture() const;

private:
    void createDepthTexture();
    void createShadowFBO();
    void updateDepthTexture();
    void buildDDepthTexture();

    // Debug
    void saveDepthBuffer();

    ShadowMapData _shadowData;

    // Texture coords in [0, 1], while clip coords in [-1, 1]
    const glm::dmat4 _toTextureCoordsMatrix = glm::dmat4(
        glm::dvec4(0.5, 0.0, 0.0, 0.0),
        glm::dvec4(0.0, 0.5, 0.0, 0.0),
        glm::dvec4(0.0, 0.0, 1.0, 0.0),
        glm::dvec4(0.5, 0.5, 0.0, 1.0)
    );

    // DEBUG
    properties::TriggerProperty _saveDepthTexture;
    properties::IntProperty _distanceFraction;
    properties::BoolProperty _enabled;

    int _shadowDepthTextureHeight = 4096;
    int _shadowDepthTextureWidth = 4096;
    bool _dynamicDepthTextureRes = true;

    GLuint _shadowDepthTexture = 0;
    // JCC: Used to avoid recompilation of Globebrowsing's shaders
    // and incorrect binding of texture type
    GLuint _dDepthTexture = 0;
    GLuint _positionInLightSpaceTexture = 0;
    GLuint _shadowFBO = 0;
    GLint _currentFBO = 0;
    GLint _mViewport[4];

    GLboolean _faceCulling;
    GLboolean _polygonOffSet;
    GLboolean _depthIsEnabled;
    GLboolean _blendIsEnabled = false;

    glm::vec3 _sunPosition = glm::vec3(0.f);

    glm::dmat4 _shadowMatrix = glm::dmat4(1.0);

    glm::dvec3 _cameraPos = glm::dvec3(0.0);
    glm::dvec3 _cameraFocus = glm::dvec3(0.0);
    glm::dquat _cameraRotation = glm::dquat(1.0, 0.0, 0.0, 0.0);

    std::stringstream _serializedCamera;

    std::unique_ptr<Camera> _lightCamera;

    // DEBUG
    bool _executeDepthTextureSave = false;
};

} // namespace openspace

#endif // __OPENSPACE_MODULE_GLOBEBROWSING___SHADOWCOMPONENT___H__

redpajama_set_name: RedPajamaGithub


text:
select member.MEMBER_ID
     , member.MEMBER_NAME
     , member.REGISTER_DATETIME
     , member.REGISTER_USER
     , member.UPDATE_DATETIME
     , member.UPDATE_USER
  from MEMBER member
 order by member.MEMBER_ID asc


text:
/**
 * 下载worker
 */
var request = require('request');
var cheerio = require('cheerio');
var fs = require('fs-extra');
var urlencode = require('urlencode');

module.exports = function(data, callback) {
	var page = data.index;
	var url = data.url;
	console.log('start download: ' + url);
	request(url, function(error, response, body) {
		if (!error && response.statusCode == 200) {
			// console.log(body);
			var $doc = cheerio.load(body);
			var $links = $doc('a');
			$links = $links.filter(function(index, link) {
				return link.attribs.href.indexOf('ed2k') >= 0;
			});
			var data = '## 皮皮书屋电驴下载资源 \n\n';
			$links.map(function(index, link) {
				var href = link.attribs.href;
				var splits = href.split('|');
				data += '![' + urlencode.decode(splits[2]) + '](' + href + ')\n\n';
				return link.attribs.href;
			});
			var file = './data/' + page + '.md';
			fs.outputFile(file, data, function(err) {
				if (err) {
					console.log('save error', err);
				}
			});
			// console.log(data);
			callback(null, 'ok');
		}
	});
}


text:
package org.switchyard.quickstarts.bean.service;

public interface InventoryService {

    Item lookupItem(String itemId) throws ItemNotFoundException;
}


text:
@(tour: Tournament)(implicit ctx: Context)

        @tour.clock.show •
        @if(tour.variant.exotic) {
        @game.variantLink(tour.variant, (if (tour.variant == chess.variant.KingOfTheHill) tour.variant.shortName else tour.variant.name), cssClass = "hint--top")
        } else {
        @tour.perfType.map(_.name)
        }
        @if(!tour.position.initial) {
        • thematic
        } • @tour.durationString

      @systemName(tour.system).capitalize
      (?)

  @if(tour.createdBy == "lichess") {
  Official
  } else {
  @trans.by(usernameOrId(tour.createdBy))
  }
  • @momentFormat(tour.startsAt)
  @if(!tour.position.initial) {

    @tour.position.eco @tour.position.name

  }
  @tour.winnerId.map { userId =>

  @trans.winner(): @userIdLink(userId.some)
  }


text:
package com.twitter.summingbird.storm.builder

import com.twitter.algebird.Semigroup
import com.twitter.algebird.util.summer.Incrementor
import com.twitter.summingbird.online.executor.{ InputState, KeyValueShards, OperationContainer }
import com.twitter.summingbird.online.option.{ MaxEmitPerExecute, SummerBuilder }
import com.twitter.summingbird.option.JobId
import com.twitter.summingbird.storm.StormMetric
import com.twitter.summingbird.storm.option.{ AckOnEntry, AnchorTuples, MaxExecutePerSecond }
import com.twitter.tormenta.spout.{ Spout => TormentaSpout }
import org.apache.storm.generated.StormTopology
import org.apache.storm.topology.{ IRichBolt, TopologyBuilder }
import org.apache.storm.{ Config => BacktypeStormConfig }
import org.apache.storm.tuple.Tuple
import scala.collection.{ Map => CMap }
import scala.util.Try

private[summingbird] case class Topology(
  spouts: Map[Topology.SpoutId[_], Topology.Spout[_]],
  bolts: Map[Topology.BoltId[_, _], Topology.Bolt[_, _]],
  edges: List[Topology.Edge[_, _]]
) {
  def withSpout[O](id: String, spout: Topology.Spout[O]): (Topology.SpoutId[O], Topology) = {
    val spoutId = Topology.SpoutId[O](id)
    assert(!spouts.contains(spoutId))
    (spoutId, Topology(spouts.updated(spoutId, spout), bolts, edges))
  }

  def withBolt[I, O](id: String, bolt: Topology.Bolt[I, O]): (Topology.BoltId[I, O], Topology) = {
    val boltId = Topology.BoltId[I, O](id)
    assert(!bolts.contains(boltId))
    (boltId, Topology(spouts, bolts.updated(boltId, bolt), edges))
  }

  def withComponent[O](id: String, component: Topology.Component[O]): (Topology.EmittingId[O], Topology) = {
    component match {
      case bolt: Topology.Bolt[_, O] => withBolt(id, bolt)
      case spout: Topology.Spout[O] => withSpout(id, spout)
    }
  }

  def withEdge[I, O](edge: Topology.Edge[I, O]): Topology = {
    assert(edge.source != edge.dest)
    assert(edges.forall { !edge.sameEndPoints(_) })

    Topology(spouts, bolts, edges :+ edge)
  }

  def contains(id: Topology.ComponentId): Boolean = id match {
    case spoutId: Topology.SpoutId[_] => spouts.contains(spoutId)
    case boltId: Topology.BoltId[_, _] => bolts.contains(boltId)
  }

  def incomingEdges[O](id: Topology.ReceivingId[O]): List[Topology.Edge[_, O]] = {
    assert(contains(id))
    edges.filter(_.dest == id).asInstanceOf[List[Topology.Edge[_, O]]]
  }

  def outgoingEdges[I](id: Topology.EmittingId[I]): List[Topology.Edge[I, _]] = {
    assert(contains(id))
    edges.filter(_.source == id).asInstanceOf[List[Topology.Edge[I, _]]]
  }

  def build(jobId: JobId): StormTopology = {
    edges.foreach(edge => assert(contains(edge.source) && contains(edge.dest)))

    val builder = new TopologyBuilder

    spouts.foreach { case (spoutId: Topology.SpoutId[Any], spout: Topology.Spout[Any]) =>
      val builtSpout = SpoutBuilder.build[Any](
        jobId,
        spoutId,
        spout,
        outgoingEdges(spoutId)
      )
      builder.setSpout(spoutId.id, builtSpout, spout.parallelism)
    }

    bolts.foreach { case (boltId: Topology.BoltId[Any, Any], bolt: Topology.Bolt[Any, Any]) =>
      val builtBolt: IRichBolt = BaseBolt[Any, Any](
        jobId,
        boltId,
        bolt.metrics,
        bolt.anchorTuples,
        bolt.ackOnEntry,
        bolt.maxExecutePerSec,
        incomingEdges(boltId).toVector,
        outgoingEdges(boltId).toVector,
        bolt.executor
      )

      val declarer = builder.setBolt(
        boltId.id,
        builtBolt,
        bolt.parallelism
      ).addConfigurations(tickConfig)

      incomingEdges(boltId).foreach { edge =>
        edge.grouping.register(declarer, edge.source)
      }
    }

    builder.createTopology()
  }

  /**
   * Set storm to tick our nodes every second to clean up finished futures
   */
  private def tickConfig = {
    val boltConfig = new BacktypeStormConfig
    boltConfig.put(BacktypeStormConfig.TOPOLOGY_TICK_TUPLE_FREQ_SECS, java.lang.Integer.valueOf(1))
    boltConfig
  }
}

private[summingbird] object Topology {
  val empty = Topology(Map(), Map(), List())

  /**
   * Represents id of topology's component.
   */
  sealed trait ComponentId {
    def id: String
  }

  /**
   * Represents id of topology's component which receives tuples.
   * @tparam I represents type of received tuples by this component.
   */
  trait ReceivingId[-I] extends ComponentId

  /**
   * Represents id of topology's component which emits tuples.
   * @tparam O represents type of emitted tuples by this component.
   */
  trait EmittingId[+O] extends ComponentId

  /**
   * Represents id of topology's spout.
   * @tparam O represents type of emitted tuples by this spout.
   */
  case class SpoutId[+O](override val id: String) extends EmittingId[O]

  /**
   * Represents id of topology's bolt.
   * @tparam I represents type on received tuples by this bolt.
   * @tparam O represents type of emitted tuples by this bolt.
   */
  case class BoltId[-I, +O](override val id: String) extends ReceivingId[I] with EmittingId[O]

  /**
   * Represents topology's edge which links two components.
   * @param source id of source component.
   * @param format of tuples going through this edge.
   * @param grouping of tuples going through this edge, should be consistent with format.
   * @param onReceiveTransform will be applied on receiving side for each tuple.
   * @param dest id of destination component.
   * @tparam I type of tuples emitted by source.
   * @tparam O type of tuples received by destination.
   */
  case class Edge[I, O](
    source: EmittingId[I],
    format: OutputFormat[I],
    grouping: EdgeGrouping,
    onReceiveTransform: I => O,
    dest: ReceivingId[O]
  ) {
    def sameEndPoints(another: Edge[_, _]): Boolean =
      source == another.source && dest == another.dest

    def deserialize(serialized: java.util.List[AnyRef]): Try[O] =
      format.injection.invert(serialized).map(onReceiveTransform)
  }

  /**
   * Base trait for all components with parallelism and metrics.
   * @tparam O represent type of values emitted by this component.
   */
  sealed trait Component[+O] {
    def parallelism: Int
    def metrics: () => TraversableOnce[StormMetric[_]]
  }

  /**
   * Base trait for spouts.
   * There are two implementations: raw tormenta spout and key value spout.
   */
  sealed trait Spout[+O] extends Component[O]

  case class RawSpout[+O](
    parallelism: Int,
    metrics: () => TraversableOnce[StormMetric[_]],
    spout: TormentaSpout[O]
  ) extends Spout[O]

  case class KeyValueSpout[K, V: Semigroup](
    parallelism: Int,
    metrics: () => TraversableOnce[StormMetric[_]],
    spout: TormentaSpout[(K, V)],
    shards: KeyValueShards,
    summerBuilder: SummerBuilder,
    maxEmitPerExec: MaxEmitPerExecute,
    flushExecTimeCounter: Incrementor,
    executeTimeCounter: Incrementor
  ) extends Spout[(Int, CMap[K, V])] {
    val semigroup: Semigroup[V] = implicitly[Semigroup[V]]
  }

  /**
   * Base class for bolts.
   */
  case class Bolt[-I, +O](
    parallelism: Int,
    metrics: () => TraversableOnce[StormMetric[_]],
    anchorTuples: AnchorTuples,
    ackOnEntry: AckOnEntry,
    maxExecutePerSec: MaxExecutePerSecond,
    executor: OperationContainer[I, O, InputState[Tuple]]
  ) extends Component[O]
}


text:
<?php declare(strict_types=1);

namespace PAF\Modules\CommissionModule\AuditTrail;

use LeanMapper\Events;
use SeStep\NetteAuditTrail\Facade\AuditTrailRepositoryAdapter;
use PAF\Modules\CommissionModule\Model\Commission;

class CommissionAuditTrailAdapter extends AuditTrailRepositoryAdapter
{

    /**
     * @inheritDoc
     */
    public function getEvents(): array
    {
        return [
            Events::EVENT_AFTER_CREATE => [$this, 'afterCreate'],
            Events::EVENT_AFTER_UPDATE => [$this, 'afterUpdate'],
        ];
    }

    public function afterCreate(Commission $commission)
    {
        $this->auditTrailService->addEvent($commission->id, 'commission.log.commissionCreated');
    }

    public function afterUpdate(Commission $commission)
    {
        $parameters['changes'] = $this->compare($commission);

        $this->auditTrailService->addEvent($commission->id, 'commission.log.commissionUpdated', $parameters);
    }
}


text:
layout: page
categories: dictionary ct2010
---

##Pedestrians Injured 1995-2013 Variables

---

####**geoid**
Unique 11-digit US 2010 Census Tract ID.

####**t10pei9513**
Pedestrians Injured 1995-2013

---


text:
using System.Collections.Generic;
using System.Net.Test.Common;
using System.Runtime.InteropServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

using Xunit;

namespace System.Net.Sockets.Tests
{
    public class SocketOptionNameTest
    {
        private static bool SocketsReuseUnicastPortSupport => Capability.SocketsReuseUnicastPortSupport().HasValue;

        [OuterLoop] // TODO: Issue #11345
        [ConditionalFact(nameof(SocketsReuseUnicastPortSupport))]
        public void ReuseUnicastPort_CreateSocketGetOption()
        {
            using (var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
            {
                if (Capability.SocketsReuseUnicastPortSupport().Value)
                {
                    Assert.Equal(0, (int)socket.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseUnicastPort));
                }
                else
                {
                    Assert.Throws<SocketException>(() => socket.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseUnicastPort));
                }
            }
        }

        [OuterLoop] // TODO: Issue #11345
        [ConditionalFact(nameof(SocketsReuseUnicastPortSupport))]
        public void ReuseUnicastPort_CreateSocketSetOption()
        {
            using (var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
            {
                if (Capability.SocketsReuseUnicastPortSupport().Value)
                {
                    socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseUnicastPort, 0);
                    int optionValue = (int)socket.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseUnicastPort);
                    Assert.Equal(0, optionValue);
                }
                else
                {
                    Assert.Throws<SocketException>(() => socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseUnicastPort, 1));
                }
            }
        }

        // TODO: Issue #4887
        // The socket option 'ReuseUnicastPost' only works on Windows 10 systems. In addition, setting the option
        // is a no-op unless specialized network settings using PowerShell configuration are first applied to the
        // machine. This is currently difficult to test in the CI environment. So, this test will be disabled for now
        [OuterLoop] // TODO: Issue #11345
        [ActiveIssue(4887)]
        public void ReuseUnicastPort_CreateSocketSetOptionToOneAndGetOption_SocketsReuseUnicastPortSupport_OptionIsOne()
        {
            var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);

            socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseUnicastPort, 1);
            int optionValue = (int)socket.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseUnicastPort);
            Assert.Equal(1, optionValue);
        }

        [OuterLoop] // TODO: Issue #11345
        [Fact]
        public void MulticastOption_CreateSocketSetGetOption_GroupAndInterfaceIndex_SetSucceeds_GetThrows()
        {
            int interfaceIndex = 0;
            IPAddress groupIp = IPAddress.Parse("239.1.2.3");

            using (Socket socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp))
            {
                socket.SetSocketOption(SocketOptionLevel.IP, SocketOptionName.AddMembership, new MulticastOption(groupIp, interfaceIndex));

                Assert.Throws<SocketException>(() => socket.GetSocketOption(SocketOptionLevel.IP, SocketOptionName.AddMembership));
            }
        }

        [OuterLoop] // TODO: Issue #11345
        [Fact]
        public async Task MulticastInterface_Set_AnyInterface_Succeeds()
        {
            // On all platforms, index 0 means "any interface"
            await MulticastInterface_Set_Helper(0);
        }

        [OuterLoop] // TODO: Issue #11345
        [Fact]
        [PlatformSpecific(TestPlatforms.Windows)] // see comment below
        [ActiveIssue(21327, TargetFrameworkMonikers.Uap)] // UWP Apps are forbidden to send network traffic to the local Computer.
        public async Task MulticastInterface_Set_Loopback_Succeeds()
        {
            // On Windows, we can apparently assume interface 1 is "loopback." On other platforms, this is not a
            // valid assumption. We could maybe use NetworkInterface.LoopbackInterfaceIndex to get the index, but
            // this would introduce a dependency on System.Net.NetworkInformation, which depends on System.Net.Sockets,
            // which is what we're testing here....  So for now, we'll just assume "loopback == 1" and run this on
            // Windows only.
            await MulticastInterface_Set_Helper(1);
        }

        private async Task MulticastInterface_Set_Helper(int interfaceIndex)
        {
            IPAddress multicastAddress = IPAddress.Parse("239.1.2.3");
            string message = "hello";
            int port;

            using (Socket receiveSocket = CreateBoundUdpSocket(out port),
                          sendSocket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp))
            {
                receiveSocket.ReceiveTimeout = 1000;
                receiveSocket.SetSocketOption(SocketOptionLevel.IP, SocketOptionName.AddMembership, new MulticastOption(multicastAddress, interfaceIndex));

                sendSocket.SetSocketOption(SocketOptionLevel.IP, SocketOptionName.MulticastInterface, IPAddress.HostToNetworkOrder(interfaceIndex));

                var receiveBuffer = new byte[1024];
                var receiveTask = receiveSocket.ReceiveAsync(new ArraySegment<byte>(receiveBuffer), SocketFlags.None);

                for (int i = 0; i < TestSettings.UDPRedundancy; i++)
                {
                    sendSocket.SendTo(Encoding.UTF8.GetBytes(message), new IPEndPoint(multicastAddress, port));
                }

                var cts = new CancellationTokenSource();
                Assert.True(await Task.WhenAny(receiveTask, Task.Delay(20_000, cts.Token)) == receiveTask, "Waiting for received data timed out");
                cts.Cancel();

                int bytesReceived = await receiveTask;
                string receivedMessage = Encoding.UTF8.GetString(receiveBuffer, 0, bytesReceived);

                Assert.Equal(receivedMessage, message);
            }
        }

        [OuterLoop] // TODO: Issue #11345
        [Fact]
        public void MulticastInterface_Set_InvalidIndex_Throws()
        {
            int interfaceIndex = 31415;
            using (Socket s = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp))
            {
                Assert.Throws<SocketException>(() =>
                    s.SetSocketOption(SocketOptionLevel.IP, SocketOptionName.MulticastInterface, IPAddress.HostToNetworkOrder(interfaceIndex)));
            }
        }

        [OuterLoop] // TODO: Issue #11345
        [ConditionalTheory(typeof(PlatformDetection), nameof(PlatformDetection.IsNotWindowsSubsystemForLinux))] // In WSL, the connect() call fails immediately.
        [InlineData(false)]
        [InlineData(true)]
        public void FailedConnect_GetSocketOption_SocketOptionNameError(bool simpleGet)
        {
            using (var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp) { Blocking = false })
            {
                // Fail a Connect
                using (var server = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
                {
                    server.Bind(new IPEndPoint(IPAddress.Loopback, 0)); // bind but don't listen
                    Assert.ThrowsAny<Exception>(() => client.Connect(server.LocalEndPoint));
                }

                // Verify via Select that there's an error
                const int FailedTimeout = 10 * 1000 * 1000; // 10 seconds
                var errorList = new List<Socket> { client };
                Socket.Select(null, null, errorList, FailedTimeout);
                Assert.Equal(1, errorList.Count);

                // Get the last error and validate it's what's expected
                int errorCode;
                if (simpleGet)
                {
                    errorCode = (int)client.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Error);
                }
                else
                {
                    byte[] optionValue = new byte[sizeof(int)];
                    client.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Error, optionValue);
                    errorCode = BitConverter.ToInt32(optionValue, 0);
                }
                Assert.Equal((int)SocketError.ConnectionRefused, errorCode);

                // Then get it again
                if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
                {
                    // The Windows implementation doesn't clear the error code after retrieved.
                    // https://github.com/dotnet/corefx/issues/8464
                    Assert.Equal(errorCode, (int)client.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Error));
                }
                else
                {
                    // The Unix implementation matches the getsockopt and MSDN docs and clears the error code as part of retrieval.
                    Assert.Equal((int)SocketError.Success, (int)client.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Error));
                }
            }
        }

        // Create an Udp Socket and binds it to an available port
        private static Socket CreateBoundUdpSocket(out int localPort)
        {
            Socket receiveSocket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);

            // sending a message will bind the socket to an available port
            string sendMessage = "dummy message";
            int port = 54320;
            IPAddress multicastAddress = IPAddress.Parse("239.1.1.1");
            receiveSocket.SendTo(Encoding.UTF8.GetBytes(sendMessage), new IPEndPoint(multicastAddress, port));

            localPort = (receiveSocket.LocalEndPoint as IPEndPoint).Port;
            return receiveSocket;
        }

        [Theory]
        [InlineData(null, null, null, true)]
        [InlineData(null, null, false, true)]
        [InlineData(null, false, false, true)]
        [InlineData(null, true, false, true)]
        [InlineData(null, true, true, false)]
        [InlineData(true, null, null, true)]
        [InlineData(true, null, false, true)]
        [InlineData(true, null, true, true)]
        [InlineData(true, false, null, true)]
        [InlineData(true, false, false, true)]
        [InlineData(true, false, true, true)]
        public void ReuseAddress(bool? exclusiveAddressUse, bool? firstSocketReuseAddress, bool? secondSocketReuseAddress, bool expectFailure)
        {
            using (Socket a = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp))
            {
                if (exclusiveAddressUse.HasValue)
                {
                    a.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ExclusiveAddressUse, exclusiveAddressUse.Value);
                }
                if (firstSocketReuseAddress.HasValue)
                {
                    a.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, firstSocketReuseAddress.Value);
                }

                a.Bind(new IPEndPoint(IPAddress.Loopback, 0));

                using (Socket b = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp))
                {
                    if (secondSocketReuseAddress.HasValue)
                    {
                        b.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, secondSocketReuseAddress.Value);
                    }

                    if (expectFailure)
                    {
                        Assert.ThrowsAny<SocketException>(() => b.Bind(a.LocalEndPoint));
                    }
                    else
                    {
                        b.Bind(a.LocalEndPoint);
                    }
                }
            }
        }

        [Theory]
        [PlatformSpecific(TestPlatforms.Windows)]  // ExclusiveAddressUse option is a Windows-specific option (when set to "true," tells Windows not to allow reuse of same local address)
        [InlineData(false, null, null, true)]
        [InlineData(false, null, false, true)]
        [InlineData(false, false, null, true)]
        [InlineData(false, false, false, true)]
        [InlineData(false, true, null, true)]
        [InlineData(false, true, false, true)]
        [InlineData(false, true, true, false)]
        public void ReuseAddress_Windows(bool? exclusiveAddressUse, bool? firstSocketReuseAddress, bool? secondSocketReuseAddress, bool expectFailure)
        {
            ReuseAddress(exclusiveAddressUse, firstSocketReuseAddress, secondSocketReuseAddress, expectFailure);
        }

        [Fact]
        [PlatformSpecific(TestPlatforms.AnyUnix)] // Windows defaults are different
        public void ExclusiveAddress_Default_Unix()
        {
            using (Socket a = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp))
            {
                Assert.Equal(1, (int)a.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ExclusiveAddressUse));
                Assert.Equal(true, a.ExclusiveAddressUse);
                Assert.Equal(0, (int)a.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress));
            }
        }

        [Theory]
        [InlineData(1)]
        [InlineData(0)]
        [PlatformSpecific(TestPlatforms.AnyUnix)] // Unix does not have separate options for ExclusiveAddressUse and ReuseAddress.
        public void SettingExclusiveAddress_SetsReuseAddress(int value)
        {
            using (Socket a = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp))
            {
                a.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ExclusiveAddressUse, value);

                Assert.Equal(value, (int)a.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ExclusiveAddressUse));
                Assert.Equal(value == 1 ? 0 : 1, (int)a.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress));
            }

            // SettingReuseAddress_SetsExclusiveAddress
            using (Socket a = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp))
            {
                a.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, value);

                Assert.Equal(value, (int)a.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress));
                Assert.Equal(value == 1 ? 0 : 1, (int)a.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ExclusiveAddressUse));
            }
        }

        [Fact]
        public void BindDuringTcpWait_Succeeds()
        {
            int port = 0;
            using (Socket a = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
            {
                a.Bind(new IPEndPoint(IPAddress.Loopback, 0));
                port = (a.LocalEndPoint as IPEndPoint).Port;
                a.Listen(10);

                // Connect a client
                using (Socket client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
                {
                    client.Connect(new IPEndPoint(IPAddress.Loopback, port));
                    // accept socket and close it with zero linger time.
                    a.Accept().Close(0);
                }
            }

            // Bind a socket to the same address we just used.
            using (Socket b = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
            {
                b.Bind(new IPEndPoint(IPAddress.Loopback, port));
            }
        }

        [Fact]
        public void ExclusiveAddressUseTcp()
        {
            using (Socket a = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
            {
                // ExclusiveAddressUse defaults to true on Unix, on Windows it defaults to false.
                a.ExclusiveAddressUse = true;

                a.Bind(new IPEndPoint(IPAddress.Loopback, 0));
                a.Listen(10);
                int port = (a.LocalEndPoint as IPEndPoint).Port;

                using (Socket b = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
                {
                    SocketException ex = Assert.ThrowsAny<SocketException>(() => b.Bind(new IPEndPoint(IPAddress.Loopback, port)));
                    Assert.Equal(SocketError.AddressAlreadyInUse, ex.SocketErrorCode);
                }
            }
        }

        [OuterLoop] // TODO: Issue #11345
        [Theory]
        [PlatformSpecific(TestPlatforms.Windows)]  // SetIPProtectionLevel not supported on Unix
        [InlineData(IPProtectionLevel.EdgeRestricted, AddressFamily.InterNetwork, SocketOptionLevel.IP)]
        [InlineData(IPProtectionLevel.Restricted, AddressFamily.InterNetwork, SocketOptionLevel.IP)]
        [InlineData(IPProtectionLevel.Unrestricted, AddressFamily.InterNetwork, SocketOptionLevel.IP)]
        [InlineData(IPProtectionLevel.EdgeRestricted, AddressFamily.InterNetworkV6, SocketOptionLevel.IPv6)]
        [InlineData(IPProtectionLevel.Restricted, AddressFamily.InterNetworkV6, SocketOptionLevel.IPv6)]
        [InlineData(IPProtectionLevel.Unrestricted, AddressFamily.InterNetworkV6, SocketOptionLevel.IPv6)]
        public void SetIPProtectionLevel_Windows(IPProtectionLevel level, AddressFamily family, SocketOptionLevel optionLevel)
        {
            using (var socket = new Socket(family, SocketType.Stream, ProtocolType.Tcp))
            {
                socket.SetIPProtectionLevel(level);

                int result = (int)socket.GetSocketOption(optionLevel, SocketOptionName.IPProtectionLevel);
                Assert.Equal(result, (int)level);
            }
        }

        [OuterLoop] // TODO: Issue #11345
        [Theory]
        [PlatformSpecific(TestPlatforms.AnyUnix)]  // SetIPProtectionLevel not supported on Unix
        [InlineData(IPProtectionLevel.EdgeRestricted, AddressFamily.InterNetwork)]
        [InlineData(IPProtectionLevel.Restricted, AddressFamily.InterNetwork)]
        [InlineData(IPProtectionLevel.Unrestricted, AddressFamily.InterNetwork)]
        [InlineData(IPProtectionLevel.EdgeRestricted, AddressFamily.InterNetworkV6)]
        [InlineData(IPProtectionLevel.Restricted, AddressFamily.InterNetworkV6)]
        [InlineData(IPProtectionLevel.Unrestricted, AddressFamily.InterNetworkV6)]
        public void SetIPProtectionLevel_Unix(IPProtectionLevel level, AddressFamily family)
        {
            using (var socket = new Socket(family, SocketType.Stream, ProtocolType.Tcp))
            {
                Assert.Throws<PlatformNotSupportedException>(() => socket.SetIPProtectionLevel(level));
            }
        }

        [OuterLoop] // TODO: Issue #11345
        [Theory]
        [InlineData(AddressFamily.InterNetwork)]
        [InlineData(AddressFamily.InterNetworkV6)]
        public void SetIPProtectionLevel_ArgumentException(AddressFamily family)
        {
            using (var socket = new Socket(family, SocketType.Stream, ProtocolType.Tcp))
            {
                AssertExtensions.Throws<ArgumentException>("level", () => socket.SetIPProtectionLevel(IPProtectionLevel.Unspecified));
            }
        }
    }
}


text:
package metadata

import (
	"time"

	"go.opentelemetry.io/collector/component"
	"go.opentelemetry.io/collector/confmap"
	"go.opentelemetry.io/collector/pdata/pcommon"
	"go.opentelemetry.io/collector/pdata/pmetric"
)

// MetricSettings provides common settings for a particular metric.
type MetricSettings struct {
	Enabled bool `mapstructure:"enabled"`

	enabledProvidedByUser bool
}

// IsEnabledProvidedByUser returns true if `enabled` option is explicitly set in user settings to any value.
func (ms *MetricSettings) IsEnabledProvidedByUser() bool {
	return ms.enabledProvidedByUser
}

func (ms *MetricSettings) Unmarshal(parser *confmap.Conf) error {
	if parser == nil {
		return nil
	}
	err := parser.Unmarshal(ms, confmap.WithErrorUnused())
	if err != nil {
		return err
	}
	ms.enabledProvidedByUser = parser.IsSet("enabled")
	return nil
}

// MetricsSettings provides settings for zookeeperreceiver metrics.
type MetricsSettings struct {
	ZookeeperConnectionActive            MetricSettings `mapstructure:"zookeeper.connection.active"`
	ZookeeperDataTreeEphemeralNodeCount  MetricSettings `mapstructure:"zookeeper.data_tree.ephemeral_node.count"`
	ZookeeperDataTreeSize                MetricSettings `mapstructure:"zookeeper.data_tree.size"`
	ZookeeperFileDescriptorLimit         MetricSettings `mapstructure:"zookeeper.file_descriptor.limit"`
	ZookeeperFileDescriptorOpen          MetricSettings `mapstructure:"zookeeper.file_descriptor.open"`
	ZookeeperFollowerCount               MetricSettings `mapstructure:"zookeeper.follower.count"`
	ZookeeperFsyncExceededThresholdCount MetricSettings `mapstructure:"zookeeper.fsync.exceeded_threshold.count"`
	ZookeeperLatencyAvg                  MetricSettings `mapstructure:"zookeeper.latency.avg"`
	ZookeeperLatencyMax                  MetricSettings `mapstructure:"zookeeper.latency.max"`
	ZookeeperLatencyMin                  MetricSettings `mapstructure:"zookeeper.latency.min"`
	ZookeeperPacketCount                 MetricSettings `mapstructure:"zookeeper.packet.count"`
	ZookeeperRequestActive               MetricSettings `mapstructure:"zookeeper.request.active"`
	ZookeeperSyncPending                 MetricSettings `mapstructure:"zookeeper.sync.pending"`
	ZookeeperWatchCount                  MetricSettings `mapstructure:"zookeeper.watch.count"`
	ZookeeperZnodeCount                  MetricSettings `mapstructure:"zookeeper.znode.count"`
}

func DefaultMetricsSettings() MetricsSettings {
	return MetricsSettings{
		ZookeeperConnectionActive: MetricSettings{
			Enabled: true,
		},
		ZookeeperDataTreeEphemeralNodeCount: MetricSettings{
			Enabled: true,
		},
		ZookeeperDataTreeSize: MetricSettings{
			Enabled: true,
		},
		ZookeeperFileDescriptorLimit: MetricSettings{
			Enabled: true,
		},
		ZookeeperFileDescriptorOpen: MetricSettings{
			Enabled: true,
		},
		ZookeeperFollowerCount: MetricSettings{
			Enabled: true,
		},
		ZookeeperFsyncExceededThresholdCount: MetricSettings{
			Enabled: true,
		},
		ZookeeperLatencyAvg: MetricSettings{
			Enabled: true,
		},
		ZookeeperLatencyMax: MetricSettings{
			Enabled: true,
		},
		ZookeeperLatencyMin: MetricSettings{
			Enabled: true,
		},
		ZookeeperPacketCount: MetricSettings{
			Enabled: true,
		},
		ZookeeperRequestActive: MetricSettings{
			Enabled: true,
		},
		ZookeeperSyncPending: MetricSettings{
			Enabled: true,
		},
		ZookeeperWatchCount: MetricSettings{
			Enabled: true,
		},
		ZookeeperZnodeCount: MetricSettings{
			Enabled: true,
		},
	}
}

// AttributeDirection specifies the a value direction attribute.
type AttributeDirection int

const (
	_ AttributeDirection = iota
	AttributeDirectionReceived
	AttributeDirectionSent
)

// String returns the string representation of the AttributeDirection.
func (av AttributeDirection) String() string {
	switch av {
	case AttributeDirectionReceived:
		return "received"
	case AttributeDirectionSent:
		return "sent"
	}
	return ""
}

// MapAttributeDirection is a helper map of string to AttributeDirection attribute value.
var MapAttributeDirection = map[string]AttributeDirection{
	"received": AttributeDirectionReceived,
	"sent":     AttributeDirectionSent,
}

// AttributeState specifies the a value state attribute.
type AttributeState int

const (
	_ AttributeState = iota
	AttributeStateSynced
	AttributeStateUnsynced
)

// String returns the string representation of the AttributeState.
func (av AttributeState) String() string {
	switch av {
	case AttributeStateSynced:
		return "synced"
	case AttributeStateUnsynced:
		return "unsynced"
	}
	return ""
}

// MapAttributeState is a helper map of string to AttributeState attribute value.
var MapAttributeState = map[string]AttributeState{
	"synced":   AttributeStateSynced,
	"unsynced": AttributeStateUnsynced,
}

type metricZookeeperConnectionActive struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.connection.active metric with initial data.
func (m *metricZookeeperConnectionActive) init() {
	m.data.SetName("zookeeper.connection.active")
	m.data.SetDescription("Number of active clients connected to a ZooKeeper server.")
	m.data.SetUnit("{connections}")
	m.data.SetEmptySum()
	m.data.Sum().SetIsMonotonic(false)
	m.data.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
}

func (m *metricZookeeperConnectionActive) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Sum().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperConnectionActive) updateCapacity() {
	if m.data.Sum().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Sum().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperConnectionActive) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Sum().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperConnectionActive(settings MetricSettings) metricZookeeperConnectionActive {
	m := metricZookeeperConnectionActive{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperDataTreeEphemeralNodeCount struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.data_tree.ephemeral_node.count metric with initial data.
func (m *metricZookeeperDataTreeEphemeralNodeCount) init() {
	m.data.SetName("zookeeper.data_tree.ephemeral_node.count")
	m.data.SetDescription("Number of ephemeral nodes that a ZooKeeper server has in its data tree.")
	m.data.SetUnit("{nodes}")
	m.data.SetEmptySum()
	m.data.Sum().SetIsMonotonic(false)
	m.data.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
}

func (m *metricZookeeperDataTreeEphemeralNodeCount) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Sum().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperDataTreeEphemeralNodeCount) updateCapacity() {
	if m.data.Sum().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Sum().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperDataTreeEphemeralNodeCount) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Sum().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperDataTreeEphemeralNodeCount(settings MetricSettings) metricZookeeperDataTreeEphemeralNodeCount {
	m := metricZookeeperDataTreeEphemeralNodeCount{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperDataTreeSize struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.data_tree.size metric with initial data.
func (m *metricZookeeperDataTreeSize) init() {
	m.data.SetName("zookeeper.data_tree.size")
	m.data.SetDescription("Size of data in bytes that a ZooKeeper server has in its data tree.")
	m.data.SetUnit("By")
	m.data.SetEmptySum()
	m.data.Sum().SetIsMonotonic(false)
	m.data.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
}

func (m *metricZookeeperDataTreeSize) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Sum().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperDataTreeSize) updateCapacity() {
	if m.data.Sum().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Sum().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperDataTreeSize) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Sum().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperDataTreeSize(settings MetricSettings) metricZookeeperDataTreeSize {
	m := metricZookeeperDataTreeSize{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperFileDescriptorLimit struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.file_descriptor.limit metric with initial data.
func (m *metricZookeeperFileDescriptorLimit) init() {
	m.data.SetName("zookeeper.file_descriptor.limit")
	m.data.SetDescription("Maximum number of file descriptors that a ZooKeeper server can open.")
	m.data.SetUnit("{file_descriptors}")
	m.data.SetEmptyGauge()
}

func (m *metricZookeeperFileDescriptorLimit) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Gauge().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperFileDescriptorLimit) updateCapacity() {
	if m.data.Gauge().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Gauge().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperFileDescriptorLimit) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Gauge().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperFileDescriptorLimit(settings MetricSettings) metricZookeeperFileDescriptorLimit {
	m := metricZookeeperFileDescriptorLimit{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperFileDescriptorOpen struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.file_descriptor.open metric with initial data.
func (m *metricZookeeperFileDescriptorOpen) init() {
	m.data.SetName("zookeeper.file_descriptor.open")
	m.data.SetDescription("Number of file descriptors that a ZooKeeper server has open.")
	m.data.SetUnit("{file_descriptors}")
	m.data.SetEmptySum()
	m.data.Sum().SetIsMonotonic(false)
	m.data.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
}

func (m *metricZookeeperFileDescriptorOpen) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Sum().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperFileDescriptorOpen) updateCapacity() {
	if m.data.Sum().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Sum().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperFileDescriptorOpen) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Sum().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperFileDescriptorOpen(settings MetricSettings) metricZookeeperFileDescriptorOpen {
	m := metricZookeeperFileDescriptorOpen{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperFollowerCount struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.follower.count metric with initial data.
func (m *metricZookeeperFollowerCount) init() {
	m.data.SetName("zookeeper.follower.count")
	m.data.SetDescription("The number of followers. Only exposed by the leader.")
	m.data.SetUnit("{followers}")
	m.data.SetEmptySum()
	m.data.Sum().SetIsMonotonic(false)
	m.data.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
	m.data.Sum().DataPoints().EnsureCapacity(m.capacity)
}

func (m *metricZookeeperFollowerCount) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64, stateAttributeValue string) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Sum().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
	dp.Attributes().PutStr("state", stateAttributeValue)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperFollowerCount) updateCapacity() {
	if m.data.Sum().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Sum().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperFollowerCount) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Sum().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperFollowerCount(settings MetricSettings) metricZookeeperFollowerCount {
	m := metricZookeeperFollowerCount{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperFsyncExceededThresholdCount struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.fsync.exceeded_threshold.count metric with initial data.
func (m *metricZookeeperFsyncExceededThresholdCount) init() {
	m.data.SetName("zookeeper.fsync.exceeded_threshold.count")
	m.data.SetDescription("Number of times fsync duration has exceeded warning threshold.")
	m.data.SetUnit("{events}")
	m.data.SetEmptySum()
	m.data.Sum().SetIsMonotonic(true)
	m.data.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
}

func (m *metricZookeeperFsyncExceededThresholdCount) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Sum().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperFsyncExceededThresholdCount) updateCapacity() {
	if m.data.Sum().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Sum().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperFsyncExceededThresholdCount) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Sum().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperFsyncExceededThresholdCount(settings MetricSettings) metricZookeeperFsyncExceededThresholdCount {
	m := metricZookeeperFsyncExceededThresholdCount{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperLatencyAvg struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.latency.avg metric with initial data.
func (m *metricZookeeperLatencyAvg) init() {
	m.data.SetName("zookeeper.latency.avg")
	m.data.SetDescription("Average time in milliseconds for requests to be processed.")
	m.data.SetUnit("ms")
	m.data.SetEmptyGauge()
}

func (m *metricZookeeperLatencyAvg) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Gauge().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperLatencyAvg) updateCapacity() {
	if m.data.Gauge().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Gauge().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperLatencyAvg) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Gauge().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperLatencyAvg(settings MetricSettings) metricZookeeperLatencyAvg {
	m := metricZookeeperLatencyAvg{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperLatencyMax struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.latency.max metric with initial data.
func (m *metricZookeeperLatencyMax) init() {
	m.data.SetName("zookeeper.latency.max")
	m.data.SetDescription("Maximum time in milliseconds for requests to be processed.")
	m.data.SetUnit("ms")
	m.data.SetEmptyGauge()
}

func (m *metricZookeeperLatencyMax) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Gauge().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperLatencyMax) updateCapacity() {
	if m.data.Gauge().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Gauge().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperLatencyMax) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Gauge().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperLatencyMax(settings MetricSettings) metricZookeeperLatencyMax {
	m := metricZookeeperLatencyMax{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperLatencyMin struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.latency.min metric with initial data.
func (m *metricZookeeperLatencyMin) init() {
	m.data.SetName("zookeeper.latency.min")
	m.data.SetDescription("Minimum time in milliseconds for requests to be processed.")
	m.data.SetUnit("ms")
	m.data.SetEmptyGauge()
}

func (m *metricZookeeperLatencyMin) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Gauge().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperLatencyMin) updateCapacity() {
	if m.data.Gauge().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Gauge().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperLatencyMin) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Gauge().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperLatencyMin(settings MetricSettings) metricZookeeperLatencyMin {
	m := metricZookeeperLatencyMin{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperPacketCount struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.packet.count metric with initial data.
func (m *metricZookeeperPacketCount) init() {
	m.data.SetName("zookeeper.packet.count")
	m.data.SetDescription("The number of ZooKeeper packets received or sent by a server.")
	m.data.SetUnit("{packets}")
	m.data.SetEmptySum()
	m.data.Sum().SetIsMonotonic(true)
	m.data.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
	m.data.Sum().DataPoints().EnsureCapacity(m.capacity)
}

func (m *metricZookeeperPacketCount) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64, directionAttributeValue string) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Sum().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
	dp.Attributes().PutStr("direction", directionAttributeValue)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperPacketCount) updateCapacity() {
	if m.data.Sum().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Sum().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperPacketCount) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Sum().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperPacketCount(settings MetricSettings) metricZookeeperPacketCount {
	m := metricZookeeperPacketCount{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperRequestActive struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.request.active metric with initial data.
func (m *metricZookeeperRequestActive) init() {
	m.data.SetName("zookeeper.request.active")
	m.data.SetDescription("Number of currently executing requests.")
	m.data.SetUnit("{requests}")
	m.data.SetEmptySum()
	m.data.Sum().SetIsMonotonic(false)
	m.data.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
}

func (m *metricZookeeperRequestActive) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Sum().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperRequestActive) updateCapacity() {
	if m.data.Sum().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Sum().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperRequestActive) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Sum().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperRequestActive(settings MetricSettings) metricZookeeperRequestActive {
	m := metricZookeeperRequestActive{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperSyncPending struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.sync.pending metric with initial data.
func (m *metricZookeeperSyncPending) init() {
	m.data.SetName("zookeeper.sync.pending")
	m.data.SetDescription("The number of pending syncs from the followers. Only exposed by the leader.")
	m.data.SetUnit("{syncs}")
	m.data.SetEmptySum()
	m.data.Sum().SetIsMonotonic(false)
	m.data.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
}

func (m *metricZookeeperSyncPending) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Sum().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperSyncPending) updateCapacity() {
	if m.data.Sum().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Sum().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperSyncPending) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Sum().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperSyncPending(settings MetricSettings) metricZookeeperSyncPending {
	m := metricZookeeperSyncPending{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperWatchCount struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.watch.count metric with initial data.
func (m *metricZookeeperWatchCount) init() {
	m.data.SetName("zookeeper.watch.count")
	m.data.SetDescription("Number of watches placed on Z-Nodes on a ZooKeeper server.")
	m.data.SetUnit("{watches}")
	m.data.SetEmptySum()
	m.data.Sum().SetIsMonotonic(false)
	m.data.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
}

func (m *metricZookeeperWatchCount) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Sum().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperWatchCount) updateCapacity() {
	if m.data.Sum().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Sum().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperWatchCount) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Sum().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperWatchCount(settings MetricSettings) metricZookeeperWatchCount {
	m := metricZookeeperWatchCount{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

type metricZookeeperZnodeCount struct {
	data     pmetric.Metric // data buffer for generated metric.
	settings MetricSettings // metric settings provided by user.
	capacity int            // max observed number of data points added to the metric.
}

// init fills zookeeper.znode.count metric with initial data.
func (m *metricZookeeperZnodeCount) init() {
	m.data.SetName("zookeeper.znode.count")
	m.data.SetDescription("Number of z-nodes that a ZooKeeper server has in its data tree.")
	m.data.SetUnit("{znodes}")
	m.data.SetEmptySum()
	m.data.Sum().SetIsMonotonic(false)
	m.data.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
}

func (m *metricZookeeperZnodeCount) recordDataPoint(start pcommon.Timestamp, ts pcommon.Timestamp, val int64) {
	if !m.settings.Enabled {
		return
	}
	dp := m.data.Sum().DataPoints().AppendEmpty()
	dp.SetStartTimestamp(start)
	dp.SetTimestamp(ts)
	dp.SetIntValue(val)
}

// updateCapacity saves max length of data point slices that will be used for the slice capacity.
func (m *metricZookeeperZnodeCount) updateCapacity() {
	if m.data.Sum().DataPoints().Len() > m.capacity {
		m.capacity = m.data.Sum().DataPoints().Len()
	}
}

// emit appends recorded metric data to a metrics slice and prepares it for recording another set of data points.
func (m *metricZookeeperZnodeCount) emit(metrics pmetric.MetricSlice) {
	if m.settings.Enabled && m.data.Sum().DataPoints().Len() > 0 {
		m.updateCapacity()
		m.data.MoveTo(metrics.AppendEmpty())
		m.init()
	}
}

func newMetricZookeeperZnodeCount(settings MetricSettings) metricZookeeperZnodeCount {
	m := metricZookeeperZnodeCount{settings: settings}
	if settings.Enabled {
		m.data = pmetric.NewMetric()
		m.init()
	}
	return m
}

// MetricsBuilder provides an interface for scrapers to report metrics while taking care of all the transformations
// required to produce metric representation defined in metadata and user settings.
type MetricsBuilder struct {
	startTime                                  pcommon.Timestamp   // start time that will be applied to all recorded data points.
	metricsCapacity                            int                 // maximum observed number of metrics per resource.
	resourceCapacity                           int                 // maximum observed number of resource attributes.
	metricsBuffer                              pmetric.Metrics     // accumulates metrics data before emitting.
	buildInfo                                  component.BuildInfo // contains version information
	metricZookeeperConnectionActive            metricZookeeperConnectionActive
	metricZookeeperDataTreeEphemeralNodeCount  metricZookeeperDataTreeEphemeralNodeCount
	metricZookeeperDataTreeSize                metricZookeeperDataTreeSize
	metricZookeeperFileDescriptorLimit         metricZookeeperFileDescriptorLimit
	metricZookeeperFileDescriptorOpen          metricZookeeperFileDescriptorOpen
	metricZookeeperFollowerCount               metricZookeeperFollowerCount
	metricZookeeperFsyncExceededThresholdCount metricZookeeperFsyncExceededThresholdCount
	metricZookeeperLatencyAvg                  metricZookeeperLatencyAvg
	metricZookeeperLatencyMax                  metricZookeeperLatencyMax
	metricZookeeperLatencyMin                  metricZookeeperLatencyMin
	metricZookeeperPacketCount                 metricZookeeperPacketCount
	metricZookeeperRequestActive               metricZookeeperRequestActive
	metricZookeeperSyncPending                 metricZookeeperSyncPending
	metricZookeeperWatchCount                  metricZookeeperWatchCount
	metricZookeeperZnodeCount                  metricZookeeperZnodeCount
}

// metricBuilderOption applies changes to default metrics builder.
type metricBuilderOption func(*MetricsBuilder)

// WithStartTime sets startTime on the metrics builder.
func WithStartTime(startTime pcommon.Timestamp) metricBuilderOption {
	return func(mb *MetricsBuilder) {
		mb.startTime = startTime
	}
}

func NewMetricsBuilder(settings MetricsSettings, buildInfo component.BuildInfo, options ...metricBuilderOption) *MetricsBuilder {
	mb := &MetricsBuilder{
		startTime:                       pcommon.NewTimestampFromTime(time.Now()),
		metricsBuffer:                   pmetric.NewMetrics(),
		buildInfo:                       buildInfo,
		metricZookeeperConnectionActive: newMetricZookeeperConnectionActive(settings.ZookeeperConnectionActive),
		metricZookeeperDataTreeEphemeralNodeCount:  newMetricZookeeperDataTreeEphemeralNodeCount(settings.ZookeeperDataTreeEphemeralNodeCount),
		metricZookeeperDataTreeSize:                newMetricZookeeperDataTreeSize(settings.ZookeeperDataTreeSize),
		metricZookeeperFileDescriptorLimit:         newMetricZookeeperFileDescriptorLimit(settings.ZookeeperFileDescriptorLimit),
		metricZookeeperFileDescriptorOpen:          newMetricZookeeperFileDescriptorOpen(settings.ZookeeperFileDescriptorOpen),
		metricZookeeperFollowerCount:               newMetricZookeeperFollowerCount(settings.ZookeeperFollowerCount),
		metricZookeeperFsyncExceededThresholdCount: newMetricZookeeperFsyncExceededThresholdCount(settings.ZookeeperFsyncExceededThresholdCount),
		metricZookeeperLatencyAvg:                  newMetricZookeeperLatencyAvg(settings.ZookeeperLatencyAvg),
		metricZookeeperLatencyMax:                  newMetricZookeeperLatencyMax(settings.ZookeeperLatencyMax),
		metricZookeeperLatencyMin:                  newMetricZookeeperLatencyMin(settings.ZookeeperLatencyMin),
		metricZookeeperPacketCount:                 newMetricZookeeperPacketCount(settings.ZookeeperPacketCount),
		metricZookeeperRequestActive:               newMetricZookeeperRequestActive(settings.ZookeeperRequestActive),
		metricZookeeperSyncPending:                 newMetricZookeeperSyncPending(settings.ZookeeperSyncPending),
		metricZookeeperWatchCount:                  newMetricZookeeperWatchCount(settings.ZookeeperWatchCount),
		metricZookeeperZnodeCount:                  newMetricZookeeperZnodeCount(settings.ZookeeperZnodeCount),
	}
	for _, op := range options {
		op(mb)
	}
	return mb
}

// updateCapacity updates max length of metrics and resource attributes that will be used for the slice capacity.
func (mb *MetricsBuilder) updateCapacity(rm pmetric.ResourceMetrics) {
	if mb.metricsCapacity < rm.ScopeMetrics().At(0).Metrics().Len() {
		mb.metricsCapacity = rm.ScopeMetrics().At(0).Metrics().Len()
	}
	if mb.resourceCapacity < rm.Resource().Attributes().Len() {
		mb.resourceCapacity = rm.Resource().Attributes().Len()
	}
}

// ResourceMetricsOption applies changes to provided resource metrics.
type ResourceMetricsOption func(pmetric.ResourceMetrics)

// WithServerState sets provided value as "server.state" attribute for current resource.
func WithServerState(val string) ResourceMetricsOption {
	return func(rm pmetric.ResourceMetrics) {
		rm.Resource().Attributes().PutStr("server.state", val)
	}
}

// WithZkVersion sets provided value as "zk.version" attribute for current resource.
func WithZkVersion(val string) ResourceMetricsOption {
	return func(rm pmetric.ResourceMetrics) {
		rm.Resource().Attributes().PutStr("zk.version", val)
	}
}

// WithStartTimeOverride overrides start time for all the resource metrics data points.
// This option should be only used if different start time has to be set on metrics coming from different resources.
func WithStartTimeOverride(start pcommon.Timestamp) ResourceMetricsOption {
	return func(rm pmetric.ResourceMetrics) {
		var dps pmetric.NumberDataPointSlice
		metrics := rm.ScopeMetrics().At(0).Metrics()
		for i := 0; i < metrics.Len(); i++ {
			switch metrics.At(i).Type() {
			case pmetric.MetricTypeGauge:
				dps = metrics.At(i).Gauge().DataPoints()
			case pmetric.MetricTypeSum:
				dps = metrics.At(i).Sum().DataPoints()
			}
			for j := 0; j < dps.Len(); j++ {
				dps.At(j).SetStartTimestamp(start)
			}
		}
	}
}

// EmitForResource saves all the generated metrics under a new resource and updates the internal state to be ready for
// recording another set of data points as part of another resource. This function can be helpful when one scraper
// needs to emit metrics from several resources. Otherwise calling this function is not required,
// just `Emit` function can be called instead.
// Resource attributes should be provided as ResourceMetricsOption arguments.
func (mb *MetricsBuilder) EmitForResource(rmo ...ResourceMetricsOption) {
	rm := pmetric.NewResourceMetrics()
	rm.Resource().Attributes().EnsureCapacity(mb.resourceCapacity)
	ils := rm.ScopeMetrics().AppendEmpty()
	ils.Scope().SetName("otelcol/zookeeperreceiver")
	ils.Scope().SetVersion(mb.buildInfo.Version)
	ils.Metrics().EnsureCapacity(mb.metricsCapacity)
	mb.metricZookeeperConnectionActive.emit(ils.Metrics())
	mb.metricZookeeperDataTreeEphemeralNodeCount.emit(ils.Metrics())
	mb.metricZookeeperDataTreeSize.emit(ils.Metrics())
	mb.metricZookeeperFileDescriptorLimit.emit(ils.Metrics())
	mb.metricZookeeperFileDescriptorOpen.emit(ils.Metrics())
	mb.metricZookeeperFollowerCount.emit(ils.Metrics())
	mb.metricZookeeperFsyncExceededThresholdCount.emit(ils.Metrics())
	mb.metricZookeeperLatencyAvg.emit(ils.Metrics())
	mb.metricZookeeperLatencyMax.emit(ils.Metrics())
	mb.metricZookeeperLatencyMin.emit(ils.Metrics())
	mb.metricZookeeperPacketCount.emit(ils.Metrics())
	mb.metricZookeeperRequestActive.emit(ils.Metrics())
	mb.metricZookeeperSyncPending.emit(ils.Metrics())
	mb.metricZookeeperWatchCount.emit(ils.Metrics())
	mb.metricZookeeperZnodeCount.emit(ils.Metrics())
	for _, op := range rmo {
		op(rm)
	}
	if ils.Metrics().Len() > 0 {
		mb.updateCapacity(rm)
		rm.MoveTo(mb.metricsBuffer.ResourceMetrics().AppendEmpty())
	}
}

// Emit returns all the metrics accumulated by the metrics builder and updates the internal state to be ready for
// recording another set of metrics. This function will be responsible for applying all the transformations required to
// produce metric representation defined in metadata and user settings, e.g. delta or cumulative.
func (mb *MetricsBuilder) Emit(rmo ...ResourceMetricsOption) pmetric.Metrics {
	mb.EmitForResource(rmo...)
	metrics := pmetric.NewMetrics()
	mb.metricsBuffer.MoveTo(metrics)
	return metrics
}

// RecordZookeeperConnectionActiveDataPoint adds a data point to zookeeper.connection.active metric.
func (mb *MetricsBuilder) RecordZookeeperConnectionActiveDataPoint(ts pcommon.Timestamp, val int64) {
	mb.metricZookeeperConnectionActive.recordDataPoint(mb.startTime, ts, val)
}

// RecordZookeeperDataTreeEphemeralNodeCountDataPoint adds a data point to zookeeper.data_tree.ephemeral_node.count metric.
func (mb *MetricsBuilder) RecordZookeeperDataTreeEphemeralNodeCountDataPoint(ts pcommon.Timestamp, val int64) {
	mb.metricZookeeperDataTreeEphemeralNodeCount.recordDataPoint(mb.startTime, ts, val)
}

// RecordZookeeperDataTreeSizeDataPoint adds a data point to zookeeper.data_tree.size metric.
func (mb *MetricsBuilder) RecordZookeeperDataTreeSizeDataPoint(ts pcommon.Timestamp, val int64) {
	mb.metricZookeeperDataTreeSize.recordDataPoint(mb.startTime, ts, val)
}

// RecordZookeeperFileDescriptorLimitDataPoint adds a data point to zookeeper.file_descriptor.limit metric.
func (mb *MetricsBuilder) RecordZookeeperFileDescriptorLimitDataPoint(ts pcommon.Timestamp, val int64) {
	mb.metricZookeeperFileDescriptorLimit.recordDataPoint(mb.startTime, ts, val)
}

// RecordZookeeperFileDescriptorOpenDataPoint adds a data point to zookeeper.file_descriptor.open metric.
func (mb *MetricsBuilder) RecordZookeeperFileDescriptorOpenDataPoint(ts pcommon.Timestamp, val int64) {
	mb.metricZookeeperFileDescriptorOpen.recordDataPoint(mb.startTime, ts, val)
}

// RecordZookeeperFollowerCountDataPoint adds a data point to zookeeper.follower.count metric.
func (mb *MetricsBuilder) RecordZookeeperFollowerCountDataPoint(ts pcommon.Timestamp, val int64, stateAttributeValue AttributeState) {
	mb.metricZookeeperFollowerCount.recordDataPoint(mb.startTime, ts, val, stateAttributeValue.String())
}

// RecordZookeeperFsyncExceededThresholdCountDataPoint adds a data point to zookeeper.fsync.exceeded_threshold.count metric.
func (mb *MetricsBuilder) RecordZookeeperFsyncExceededThresholdCountDataPoint(ts pcommon.Timestamp, val int64) {
	mb.metricZookeeperFsyncExceededThresholdCount.recordDataPoint(mb.startTime, ts, val)
}

// RecordZookeeperLatencyAvgDataPoint adds a data point to zookeeper.latency.avg metric.
func (mb *MetricsBuilder) RecordZookeeperLatencyAvgDataPoint(ts pcommon.Timestamp, val int64) {
	mb.metricZookeeperLatencyAvg.recordDataPoint(mb.startTime, ts, val)
}

// RecordZookeeperLatencyMaxDataPoint adds a data point to zookeeper.latency.max metric.
func (mb *MetricsBuilder) RecordZookeeperLatencyMaxDataPoint(ts pcommon.Timestamp, val int64) {
	mb.metricZookeeperLatencyMax.recordDataPoint(mb.startTime, ts, val)
}

// RecordZookeeperLatencyMinDataPoint adds a data point to zookeeper.latency.min metric.
func (mb *MetricsBuilder) RecordZookeeperLatencyMinDataPoint(ts pcommon.Timestamp, val int64) {
	mb.metricZookeeperLatencyMin.recordDataPoint(mb.startTime, ts, val)
}

// RecordZookeeperPacketCountDataPoint adds a data point to zookeeper.packet.count metric.
func (mb *MetricsBuilder) RecordZookeeperPacketCountDataPoint(ts pcommon.Timestamp, val int64, directionAttributeValue AttributeDirection) {
	mb.metricZookeeperPacketCount.recordDataPoint(mb.startTime, ts, val, directionAttributeValue.String())
}

// RecordZookeeperRequestActiveDataPoint adds a data point to zookeeper.request.active metric.
func (mb *MetricsBuilder) RecordZookeeperRequestActiveDataPoint(ts pcommon.Timestamp, val int64) {
	mb.metricZookeeperRequestActive.recordDataPoint(mb.startTime, ts, val)
}

// RecordZookeeperSyncPendingDataPoint adds a data point to zookeeper.sync.pending metric.
func (mb *MetricsBuilder) RecordZookeeperSyncPendingDataPoint(ts pcommon.Timestamp, val int64) {
	mb.metricZookeeperSyncPending.recordDataPoint(mb.startTime, ts, val)
}

// RecordZookeeperWatchCountDataPoint adds a data point to zookeeper.watch.count metric.
func (mb *MetricsBuilder) RecordZookeeperWatchCountDataPoint(ts pcommon.Timestamp, val int64) {
	mb.metricZookeeperWatchCount.recordDataPoint(mb.startTime, ts, val)
}

// RecordZookeeperZnodeCountDataPoint adds a data point to zookeeper.znode.count metric.
func (mb *MetricsBuilder) RecordZookeeperZnodeCountDataPoint(ts pcommon.Timestamp, val int64) {
	mb.metricZookeeperZnodeCount.recordDataPoint(mb.startTime, ts, val)
}

// Reset resets metrics builder to its initial state. It should be used when external metrics source is restarted,
// and metrics builder should update its startTime and reset it's internal state accordingly.
func (mb *MetricsBuilder) Reset(options ...metricBuilderOption) {
	mb.startTime = pcommon.NewTimestampFromTime(time.Now())
	for _, op := range options {
		op(mb)
	}
}


text:
num: "lect02"
lecture_date: 2017-04-06
desc: "Review"
ready: true
pdfurl: http://andrewduncan.net/cs32/lectures/lecture_2.pdf
---

Second lecture of course.

Content is on the slides linked to above.


text:
package ch.uzh.ddis.katts.bolts.aggregate;

import static ch.uzh.ddis.katts.util.MockDataUtils.ISO_FORMAT;
import static ch.uzh.ddis.katts.util.MockDataUtils.parseString;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import javax.xml.datatype.DatatypeFactory;

import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

import ch.uzh.ddis.katts.bolts.aggregate.AggregatorManager.Callback;
import ch.uzh.ddis.katts.query.processor.aggregate.AggregatorConfiguration;
import ch.uzh.ddis.katts.query.processor.aggregate.MinAggregatorConfiguration;
import ch.uzh.ddis.katts.query.processor.aggregate.SumAggregatorConfiguration;

import com.google.common.collect.HashBasedTable;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Table;

public class AggregatorManagerTest {

	@Before
	public void setUp() throws Exception {
	}

	@After
	public void tearDown() throws Exception {
	}

	@Test
	public void testSimpleSum() throws Exception {
		AggregatorManager manager;
		AggregatorManager.Callback callback;
		DatatypeFactory datatypeFactory = DatatypeFactory.newInstance();
		SumAggregatorConfiguration sumConfig;
		List<AggregatorConfiguration<?>> aggregatorConfigList;
		ImmutableList<String> groupByKey;
		final Table<ImmutableList<Object>, String, Object> resultTable = HashBasedTable.create();

		sumConfig = new SumAggregatorConfiguration();
		sumConfig.setOf("price");
		sumConfig.setAs("total_price");
		aggregatorConfigList = new ArrayList<AggregatorConfiguration<?>>();
		aggregatorConfigList.add(sumConfig);

		callback = new Callback() {

			@Override
			public void callback(Table<ImmutableList<Object>, String, Object> aggregateValues, Date startDate,
					Date endDate) {
				synchronized (resultTable) {
					resultTable.putAll(aggregateValues);
				}
			}
		};

		manager = new AggregatorManager(datatypeFactory.newDuration("PT1S"), // window size
				datatypeFactory.newDuration("PT1S"), // update interval
				callback, // this method will be called when there are
				true, // onlyIfChanged
				sumConfig // aggregator list
		);

		groupByKey = ImmutableList.of("HHH", "sales");
		manager.incorporateValue(groupByKey,
				parseString("startDate=2001-01-01T00:00:01,ticker=HHH,department=sales,price=3.0D"));
		manager.incorporateValue(groupByKey,
				parseString("startDate=2001-01-01T00:00:02,ticker=HHH,department=sales,price=4.0D"));
		// We should have gotten the sum for HHH-sales during the first second back (written in the result table)
		synchronized (resultTable) {
			Assert.assertEquals(3.0D, ((Double) resultTable.get(groupByKey, "total_price")).doubleValue(), 0.01D);
		}
	}

	@Test
	public void testOnlyUpdateOnChangeTrue() throws Exception {
		AggregatorManager manager;
		AggregatorManager.Callback callback;
		DatatypeFactory datatypeFactory = DatatypeFactory.newInstance();
		SumAggregatorConfiguration sumConfig;
		List<AggregatorConfiguration<?>> aggregatorConfigList;
		final ImmutableList<String> groupByKey;
		final Table<ImmutableList<Object>, String, Object> resultTable = HashBasedTable.create();
		final Date[] lastUpdateDates = new Date[2]; // start- and end date
		Date lastUpdate;

		sumConfig = new SumAggregatorConfiguration();
		sumConfig.setOf("price");
		sumConfig.setAs("total_price");
		aggregatorConfigList = new ArrayList<AggregatorConfiguration<?>>();
		aggregatorConfigList.add(sumConfig);

		groupByKey = ImmutableList.of("HHH", "sales");

		callback = new Callback() {

			@Override
			public void callback(Table<ImmutableList<Object>, String, Object> aggregateValues, Date startDate,
					Date endDate) {
				synchronized (resultTable) {
					if (aggregateValues.contains(groupByKey, "total_price")) {
						resultTable.putAll(aggregateValues);
						lastUpdateDates[0] = startDate;
						lastUpdateDates[1] = endDate;
					}
				}
			}
		};

		manager = new AggregatorManager(datatypeFactory.newDuration("PT1S"), // window size
				datatypeFactory.newDuration("PT1S"), // update interval
				callback, // this method will be called when there are
				true, // onlyIfChanged
				sumConfig // aggregator list
		);

		manager.incorporateValue(groupByKey,
				parseString("startDate=2001-01-01T00:00:01,ticker=HHH,department=sales,price=3.0D"));
		manager.advanceInTime(ISO_FORMAT.parseMillis("2001-01-01T00:00:02"));
		// We should have gotten the sum for HHH-sales during the first second back (written in the result table)
		synchronized (resultTable) {
			Assert.assertEquals(3.0D, ((Double) resultTable.get(groupByKey, "total_price")).doubleValue(), 0.01D);
			lastUpdate = lastUpdateDates[1];
		}
		// Add 3.0 to the sum. since we're only looking at a 1 second window, this did not change the value
		// of the sum (it is still 3), so this should not trigger an update -> the lastUpdate should not have changed.
		manager.incorporateValue(groupByKey,
				parseString("startDate=2001-01-01T00:00:02,ticker=HHH,department=sales,price=3.0D"));
		manager.advanceInTime(ISO_FORMAT.parseMillis("2001-01-01T00:00:03"));
		synchronized (resultTable) {
			Assert.assertEquals(3.0D, ((Double) resultTable.get(groupByKey, "total_price")).doubleValue(), 0.01D);
			Assert.assertEquals(lastUpdate, lastUpdateDates[1]);
		}
	}

	@Test
	public void testOnlyUpdateOnChangeFalse() throws Exception {
		AggregatorManager manager;
		AggregatorManager.Callback callback;
		DatatypeFactory datatypeFactory = DatatypeFactory.newInstance();
		SumAggregatorConfiguration sumConfig;
		List<AggregatorConfiguration<?>> aggregatorConfigList;
		final ImmutableList<String> groupByKey;
		final Table<ImmutableList<Object>, String, Object> resultTable = HashBasedTable.create();
		final Date[] lastUpdateDates = new Date[2]; // start- and end date
		Date lastUpdate;

		sumConfig = new SumAggregatorConfiguration();
		sumConfig.setOf("price");
		sumConfig.setAs("total_price");
		aggregatorConfigList = new ArrayList<AggregatorConfiguration<?>>();
		aggregatorConfigList.add(sumConfig);

		groupByKey = ImmutableList.of("HHH", "sales");

		callback = new Callback() {

			@Override
			public void callback(Table<ImmutableList<Object>, String, Object> aggregateValues, Date startDate,
					Date endDate) {
				synchronized (resultTable) {
					if (aggregateValues.contains(groupByKey, "total_price")) {
						resultTable.putAll(aggregateValues);
						lastUpdateDates[0] = startDate;
						lastUpdateDates[1] = endDate;
					}
				}
			}
		};

		manager = new AggregatorManager(datatypeFactory.newDuration("PT1S"), // window size
				datatypeFactory.newDuration("PT1S"), // update interval
				callback, // this method will be called when there are
				false, // onlyIfChanged
				sumConfig // aggregator list
		);

		manager.incorporateValue(groupByKey,
				parseString("startDate=2001-01-01T00:00:01,ticker=HHH,department=sales,price=3.0D"));
		manager.advanceInTime(ISO_FORMAT.parseMillis("2001-01-01T00:00:02"));
		// We should have gotten the sum for HHH-sales during the first second back (written in the result table)
		synchronized (resultTable) {
			Assert.assertEquals(3.0D, ((Double) resultTable.get(groupByKey, "total_price")).doubleValue(), 0.01D);
			lastUpdate = lastUpdateDates[1];
		}
		// Add 3.0 to the sum. since we're only looking at a 1 second window, this did not change the value
		// of the sum (it is still 3), so this should not trigger an update -> the lastUpdate should not have changed.
		manager.incorporateValue(groupByKey,
				parseString("startDate=2001-01-01T00:00:03,ticker=HHH,department=sales,price=3.0D"));
		manager.advanceInTime(ISO_FORMAT.parseMillis("2001-01-01T00:00:04"));
		synchronized (resultTable) {
			Assert.assertEquals(3.0D, ((Double) resultTable.get(groupByKey, "total_price")).doubleValue(), 0.01D);
			Assert.assertFalse(lastUpdate.equals(lastUpdateDates[1]));
		}
	}

	@Test
	public void testNoWindow() throws Exception {
		AggregatorManager manager;
		AggregatorManager.Callback callback;
		DatatypeFactory datatypeFactory = DatatypeFactory.newInstance();
		SumAggregatorConfiguration sumConfig;
		List<AggregatorConfiguration<?>> aggregatorConfigList;
		final ImmutableList<String> groupByKey;
		final Table<ImmutableList<Object>, String, Object> resultTable = HashBasedTable.create();
		final Date[] lastUpdateDates = new Date[2]; // start- and end date
		Date lastUpdate;

		sumConfig = new SumAggregatorConfiguration();
		sumConfig.setOf("price");
		sumConfig.setAs("total_price");
		aggregatorConfigList = new ArrayList<AggregatorConfiguration<?>>();
		aggregatorConfigList.add(sumConfig);

		groupByKey = ImmutableList.of("HHH", "sales");

		callback = new Callback() {

			@Override
			public void callback(Table<ImmutableList<Object>, String, Object> aggregateValues, Date startDate,
					Date endDate) {
				synchronized (resultTable) {
					if (aggregateValues.contains(groupByKey, "total_price")) {
						resultTable.putAll(aggregateValues);
						lastUpdateDates[0] = startDate;
						lastUpdateDates[1] = endDate;
					}
				}
			}
		};

		manager = new AggregatorManager(null, // window size
				datatypeFactory.newDuration("PT1S"), // update interval
				callback, // this method will be called when there are
				true, // onlyIfChanged
				sumConfig // aggregator list
		);

		manager.incorporateValue(groupByKey,
				parseString("startDate=2001-01-01T00:00:01,ticker=HHH,department=sales,price=3.0D"));
		manager.advanceInTime(ISO_FORMAT.parseMillis("2001-01-01T00:00:02")); // simulate heartbeat
		// We should have gotten the sum for HHH-sales during the first second back (written in the result table)
		synchronized (resultTable) {
			Assert.assertEquals(3.0D, ((Double) resultTable.get(groupByKey, "total_price")).doubleValue(), 0.01D);
			lastUpdate = lastUpdateDates[1];
		}
		// Add 3.0 to the sum
		manager.incorporateValue(groupByKey,
				parseString("startDate=2001-01-01T00:00:02,ticker=HHH,department=sales,price=3.0D"));
		manager.advanceInTime(ISO_FORMAT.parseMillis("2001-01-01T00:00:03")); // simulate heartbeat
		// since we don't have a window, the sum should now be 6
		synchronized (resultTable) {
			Assert.assertEquals(6.0D, ((Double) resultTable.get(groupByKey, "total_price")).doubleValue(), 0.01D);
		}
	}

	@Test
	public void testMin() throws Exception {
		AggregatorManager manager;
		AggregatorManager.Callback callback;
		DatatypeFactory datatypeFactory = DatatypeFactory.newInstance();
		MinAggregatorConfiguration minConfig;
		List<AggregatorConfiguration<?>> aggregatorConfigList;
		ImmutableList<String> groupByKey;
		final Table<ImmutableList<Object>, String, Object> resultTable = HashBasedTable.create();

		minConfig = new MinAggregatorConfiguration();
		minConfig.setOf("price");
		minConfig.setAs("min_price");
		aggregatorConfigList = new ArrayList<AggregatorConfiguration<?>>();
		aggregatorConfigList.add(minConfig);

		callback = new Callback() {

			@Override
			public void callback(Table<ImmutableList<Object>, String, Object> aggregateValues, Date startDate,
					Date endDate) {
				synchronized (resultTable) {
					resultTable.putAll(aggregateValues);
				}
			}
		};

		manager = new AggregatorManager(datatypeFactory.newDuration("PT2S"), // window size
				datatypeFactory.newDuration("PT1S"), // update interval
				callback, // this method will be called when there are
				true, // onlyIfChanged
				minConfig // aggregator list
		);

		groupByKey = ImmutableList.of("HHH", "sales");
		manager.incorporateValue(groupByKey,
				parseString("startDate=2001-01-01T00:00:01,ticker=HHH,department=sales,price=3.0D"));
		manager.incorporateValue(groupByKey,
				parseString("startDate=2001-01-01T00:00:02,ticker=HHH,department=sales,price=1.0D"));
		// We should have gotten the min for HHH-sales during the first second back (written in the result table)
		synchronized (resultTable) {
			Assert.assertEquals(3.0D, ((Double) resultTable.get(groupByKey, "min_price")).doubleValue(), 0.01D);
		}
		// TODO this test is somehow croocked..
		// move one second more and now 1.0D should be the new minimum value
		manager.incorporateValue(groupByKey,
				parseString("startDate=2001-01-01T00:00:03,ticker=HHH,department=sales,price=2.0D"));
		synchronized (resultTable) {
			Assert.assertEquals(1.0D, ((Double) resultTable.get(groupByKey, "min_price")).doubleValue(), 0.01D);
		}
	}

}


text:
package nl.weeaboo.styledtext.gdx;

import org.junit.Before;
import org.junit.Test;

import com.badlogic.gdx.utils.Disposable;
import com.badlogic.gdx.utils.Logger;
import com.google.common.testing.GcFinalization;
import com.google.common.testing.GcFinalization.FinalizationPredicate;

public class GdxCleanerTest {

    private static final Logger LOG = new Logger(GdxCleanerTest.class.getName());

    private final GdxCleaner cleaner = GdxCleaner.get();

    @Before
    public void before() {
        garbageCollect(0);
    }

    @Test
    public void invalidateGC() {
        createResource(1);
        garbageCollect(0); // Unreferenced resources get collected

        invalidateGCSub();

        garbageCollect(0); // alpha can now be garbage collected
    }

    private void invalidateGCSub() {
        Resource alpha = createResource(2);
        createResource(3);

        garbageCollect(1); // We're still holding a reference to alpha

        LOG.debug("This log statements prevents alpha from being garbage-collected: " + alpha);
    }

    private Resource createResource(int id) {
        Resource resource = new Resource(id);
        GdxCleaner.get().register(resource, new Cleanup(id));
        return resource;
    }

    private void garbageCollect(final int expectedSize) {
        GcFinalization.awaitDone(new FinalizationPredicate() {
            @Override
            public boolean isDone() {
                cleaner.cleanUp();
                return cleaner.size() == expectedSize;
            }
        });
    }

    private static final class Resource {

        final int id;

        Resource(int id) {
            this.id = id;
        }

        @Override
        public String toString() {
            return String.valueOf(id);
        }

    }

    private static final class Cleanup implements Disposable {

        final int id;

        Cleanup(int id) {
            this.id = id;
        }

        @Override
        public void dispose() {
            LOG.info("Cleanup: " + id);
        }

    }

}


text:
from ez_setup import use_setuptools
use_setuptools()
from setuptools import setup, find_packages

config = {
    "name": "mauto",
    "description": "mauto is a generic macro tool for Autodesk Maya.",
    "version": "0.1.0",
    "license": "The MIT License",
    "author": "Cesar Saez",
    "author_email": "cesarte@gmail.com",
    "url": "https://www.github.com/csaez/mauto",
    "packages": find_packages(exclude=["ez_setup", "tests"]),
    "package_data": {"mauto.gui": ["images/*.*"]},
    "tests_require": ["nose", "coverage", "mock"],
    "scripts": []
}

setup(**config)