Utility method that takes the topic partitions and creates the topic partition state
holders, depending on the timestamp / watermark mode.
438991
private List<KafkaTopicPartitionState<KPH>> createPartitionStateHolders(
			Map<KafkaTopicPartition, Long> partitionsToInitialOffsets,
			int timestampWatermarkMode,
			SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic,
			SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated,
			ClassLoader userCodeClassLoader) throws IOException, ClassNotFoundException {

		// CopyOnWrite as adding discovered partitions could happen in parallel
		// while different threads iterate the partitions list
		List<KafkaTopicPartitionState<KPH>> partitionStates = new CopyOnWriteArrayList<>();

		switch (timestampWatermarkMode) {
			case NO_TIMESTAMPS_WATERMARKS: {
				for (Map.Entry<KafkaTopicPartition, Long> partitionEntry : partitionsToInitialOffsets.entrySet()) {
					// create the kafka version specific partition handle
					KPH kafkaHandle = createKafkaPartitionHandle(partitionEntry.getKey());

					KafkaTopicPartitionState<KPH> partitionState =
							new KafkaTopicPartitionState<>(partitionEntry.getKey(), kafkaHandle);
					partitionState.setOffset(partitionEntry.getValue());

					partitionStates.add(partitionState);
				}

				return partitionStates;
			}

			case PERIODIC_WATERMARKS: {
				for (Map.Entry<KafkaTopicPartition, Long> partitionEntry : partitionsToInitialOffsets.entrySet()) {
					KPH kafkaHandle = createKafkaPartitionHandle(partitionEntry.getKey());

					AssignerWithPeriodicWatermarks<T> assignerInstance =
							watermarksPeriodic.deserializeValue(userCodeClassLoader);

					KafkaTopicPartitionStateWithPeriodicWatermarks<T, KPH> partitionState =
							new KafkaTopicPartitionStateWithPeriodicWatermarks<>(
									partitionEntry.getKey(),
									kafkaHandle,
									assignerInstance);

					partitionState.setOffset(partitionEntry.getValue());

					partitionStates.add(partitionState);
				}

				return partitionStates;
			}

			case PUNCTUATED_WATERMARKS: {
				for (Map.Entry<KafkaTopicPartition, Long> partitionEntry : partitionsToInitialOffsets.entrySet()) {
					KPH kafkaHandle = createKafkaPartitionHandle(partitionEntry.getKey());

					AssignerWithPunctuatedWatermarks<T> assignerInstance =
							watermarksPunctuated.deserializeValue(userCodeClassLoader);

					KafkaTopicPartitionStateWithPunctuatedWatermarks<T, KPH> partitionState =
							new KafkaTopicPartitionStateWithPunctuatedWatermarks<>(
									partitionEntry.getKey(),
									kafkaHandle,
									assignerInstance);

					partitionState.setOffset(partitionEntry.getValue());

					partitionStates.add(partitionState);
				}

				return partitionStates;
			}
			default:
				// cannot happen, add this as a guard for the future
				throw new RuntimeException();
		}
	}
1051258
def seek_to_beginning(self, topic_partition=None):
        

        if isinstance(topic_partition, TopicPartition):
            self.consumer.seek_to_beginning(topic_partition)
        else:
            raise TypeError("topic_partition must be of type TopicPartition, create it with Create TopicPartition keyword.")
769931
def mark_topic_read(self, topic, user):
        
        if not user.is_authenticated:
            return

        forum = topic.forum
        try:
            forum_track = ForumReadTrack.objects.get(forum=forum, user=user)
        except ForumReadTrack.DoesNotExist:
            forum_track = None

        if (
            forum_track is None or
            (topic.last_post_on and forum_track.mark_time < topic.last_post_on)
        ):
            topic_track, created = TopicReadTrack.objects.get_or_create(topic=topic, user=user)
            if not created:
                topic_track.save()  # mark_time filled

            # If no other topic is unread inside the considered forum, the latter should also be
            # marked as read.
            unread_topics = (
                forum.topics
                .filter(
                    Q(tracks__user=user, tracks__mark_time__lt=F('last_post_on')) |
                    Q(
                        forum__tracks__user=user, forum__tracks__mark_time__lt=F('last_post_on'),
                        tracks__isnull=True,
                    )
                )
                .exclude(id=topic.id)
            )

            forum_topic_tracks = TopicReadTrack.objects.filter(topic__forum=forum, user=user)
            if (
                not unread_topics.exists() and
                (
                    forum_track is not None or
                    forum_topic_tracks.count() == forum.topics.filter(approved=True).count()
                )
            ):
                # The topics that are marked as read inside the forum for the given user will be
                # deleted while the forum track associated with the user must be created or updated.
                # This is done only if there are as many topic tracks as approved topics in case
                # the related forum has not beem previously marked as read.
                TopicReadTrack.objects.filter(topic__forum=forum, user=user).delete()
                forum_track, _ = ForumReadTrack.objects.get_or_create(forum=forum, user=user)
                forum_track.save()

                # Update parent forum tracks
                self._update_parent_forum_tracks(forum, user)
1076585
private Map<String, SortedSet<Partition>> getZKTopicPartitionInfo(Map<Integer, Broker> allBrokers) {
        final Map<String, SortedSet<Partition>> brokerPartitionsPerTopic = new HashMap<String, SortedSet<Partition>>();
        ZkUtils.makeSurePersistentPathExists(zkClient, ZkUtils.BrokerTopicsPath);
        List<String> topics = ZkUtils.getChildrenParentMayNotExist(zkClient, ZkUtils.BrokerTopicsPath);
        for (String topic : topics) {
            // find the number of broker partitions registered for this topic
            String brokerTopicPath = ZkUtils.BrokerTopicsPath + "/" + topic;
            List<String> brokerList = ZkUtils.getChildrenParentMayNotExist(zkClient, brokerTopicPath);
            //
            final SortedSet<Partition> sortedBrokerPartitions = new TreeSet<Partition>();
            final Set<Integer> existBids = new HashSet<Integer>();
            for (String bid : brokerList) {
                final int ibid = Integer.parseInt(bid);
                final String numPath = brokerTopicPath + "/" + bid;
                final Integer numPartition = Integer.valueOf(ZkUtils.readData(zkClient, numPath));
                for (int i = 0; i < numPartition.intValue(); i++) {
                    sortedBrokerPartitions.add(new Partition(ibid, i));
                }
                existBids.add(ibid);
            }
            // add all brokers after topic created
            for(Integer bid:allBrokers.keySet()){
                if(!existBids.contains(bid)){
                    sortedBrokerPartitions.add(new Partition(bid,0));// this broker run after topic created
                }
            }
            logger.debug("Broker ids and # of partitions on each for topic: " + topic + " = " + sortedBrokerPartitions);
            brokerPartitionsPerTopic.put(topic, sortedBrokerPartitions);
        }
        return brokerPartitionsPerTopic;
    }
56684
private void prepareTopic(final Map<String, InternalTopicMetadata> topicPartitions) {
    log.debug("Starting to validate internal topics {} in partition assignor.", topicPartitions);

    // first construct the topics to make ready
    final Map<String, InternalTopicConfig> topicsToMakeReady = new HashMap<>();

    for (final InternalTopicMetadata metadata : topicPartitions.values()) {
      final InternalTopicConfig topic = metadata.config;
      final int numPartitions = metadata.numPartitions;

      if (numPartitions < 0) {
        throw new StreamsException(String.format("%sTopic [%s] number of partitions not defined", logPrefix, topic.name()));
      }

      topic.setNumberOfPartitions(numPartitions);
      topicsToMakeReady.put(topic.name(), topic);
    }

    if (!topicsToMakeReady.isEmpty()) {
      internalTopicManager.makeReady(topicsToMakeReady);
    }

    log.debug("Completed validating internal topics {} in partition assignor.", topicPartitions);
  }
912979
void topicSpaceCreated(DestinationHandler destination) throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "topicSpaceCreated", destination);

    // Synchronize around the topic space references to stop removal and
    // additions occuring at the same time. 
    synchronized (_topicSpaces)
    {
      // There are some proxies to register
      if (_topicSpaces.containsKey(destination.getUuid()))
      {
        final ArrayList topicSpacesCreated = new ArrayList();
        final ArrayList topicsCreated = new ArrayList();

        final TemporarySubscription tsubscription =
          (TemporarySubscription) _topicSpaces.get(destination.getUuid());

        // Get the list of all the subscriptions.
        final Iterator neighbourUuids = tsubscription.iMETopicList.keySet().iterator();

        // Iterate through the list of meNames to create PubSubOutputHandlers on.
        while (neighbourUuids.hasNext())
        {
          // Reset the created topicSpaces
          topicSpacesCreated.clear();

          // Reset the created topics
          topicsCreated.clear();

          // Get the next ME name.
          final SIBUuid8 neighbourUuid = (SIBUuid8) neighbourUuids.next();

          // Get the set of subscriptions that have been registered
          final Iterator subscriptions =
            ((List) tsubscription.iMETopicList.get(neighbourUuid)).iterator();

          Neighbour neighbour = getNeighbour(neighbourUuid);
          // Need a flag to indicate if this neighbour is
          // a recovered one or not.
          boolean recoveredNeighbour = (neighbour == null);
          
          // If the Neighbour wasn't found in the real list, check the recovered list as this may
          // contain it.
          if (neighbour == null)
          {
            synchronized(_recoveredNeighbours)
            {
              neighbour = (Neighbour)_recoveredNeighbours.get(neighbourUuid);
            }
          }

          // Iterate of the set of topics
          while (subscriptions.hasNext())
          {
            // Get the next topic.
            final MESubscription subscription = (MESubscription) subscriptions.next();

            // Create the proxy - this will get the create Destination
            // and attach a PubSubOutputHandler.
            final boolean proxyCreated =
              createProxy(neighbour, 
                          destination, 
                          subscription,
                          destination.getUuid(), 
                          subscription.getTopic(),
                          false);
                          
            try
            {
              // Add to MatchSpace, but only if it isn't a recovered neighbour.
              if (!recoveredNeighbour)
                subscription.eventPostCommitAdd(null);
            }
            catch (SevereMessageStoreException e)
            {
              // No FFDC code needed
              // Can never occur
            }

            // If the proxy is created, then we need to publish the event message
            // to any neighbours that are interested
            if (proxyCreated)
            {
              if (!recoveredNeighbour)
              {
                // Add the topicSpace and topic to the set of created topics.
                topicSpacesCreated.add(destination.getUuid());
                topicsCreated.add(subscription.getTopic());
              }
              else
              {
                // Add the OutputHandler to the list of known handlers.
                PubSubOutputHandler h = destination.getPubSubOutputHandler(neighbour.getUUID());
                neighbour.addPubSubOutputHandler(h);            
              }
            }
            else
            {
              // This is an error situation, as the destination now exists, 
              // so the proxy should have been created.
              final SIErrorException e =
                new SIErrorException(nls.getFormattedMessage(
                  "INTERNAL_MESSAGING_ERROR_CWSIP0001",
                  new Object[] {
                    "com.ibm.ws.sib.processor.proxyhandler.Neighbours",
                    "1:463:1.113" },
                  null));
              
              FFDCFilter.processException(
                e,
                "com.ibm.ws.sib.processor.proxyhandler.Neighbours.topicSpaceCreated",
                "1:469:1.113",
                this);
                
              if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "topicSpaceCreated", e);  
              throw e;
            }
          }

          // Send the event message to the Neighbours interested
          // Get the Neighbour that represents this meName.
          _proxyHandler.subscribeEvent(
            topicSpacesCreated,
            topicsCreated,
            neighbour.getBusId(),
            null);
        }

        // Remove the Temporary subscriptions from the list.
        _topicSpaces.remove(destination.getUuid());
      }
      else
      {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
          SibTr.debug(tc, "No proxy subscriptions to register");
      }

    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "topicSpaceCreated");
  }
454948
public List<Partition> getPartitionList(long previousWatermark) {
    if (state.getPropAsBoolean(HAS_USER_SPECIFIED_PARTITIONS)) {
      return createUserSpecifiedPartitions();
    }

    List<Partition> partitions = new ArrayList<>();

    /*
     * Use the deprecated getPartitions(long) as a helper function, avoid duplicating logic. When it can be removed, its
     * logic will be put here.
     */
    HashMap<Long, Long> partitionMap = getPartitions(previousWatermark);

    if (partitionMap.size() == 0) {
      return partitions;
    }

    if (partitionMap.size() == 1) {
      Map.Entry<Long, Long> entry = partitionMap.entrySet().iterator().next();
      Long lwm = entry.getKey();
      Long hwm = entry.getValue();

      if (lwm == hwm) {
        if (lwm != -1) { // we always allow [-1, -1] interval due to some test cases relies on this logic.
          boolean allowEqualBoundary = state.getPropAsBoolean(ALLOW_EQUAL_WATERMARK_BOUNDARY, false);
          LOG.info("Single partition with LWM = HWM and allowEqualBoundary=" + allowEqualBoundary);
          if (!allowEqualBoundary) {
            return partitions;
          }
        }
      }
    }

    /*
     * Can't use highWatermark directly, as the partitionMap may have different precision. For example, highWatermark
     * may be specified to seconds, but partitionMap could be specified to hour or date.
     */
    Long highestWatermark = Collections.max(partitionMap.values());

    for (Map.Entry<Long, Long> entry : partitionMap.entrySet()) {
      Long partitionHighWatermark = entry.getValue();
      // Apply hasUserSpecifiedHighWatermark to the last partition, which has highestWatermark
      if (partitionHighWatermark.equals(highestWatermark)) {
        partitions.add(new Partition(entry.getKey(), partitionHighWatermark, true, hasUserSpecifiedHighWatermark));
      } else {
        // The partitionHighWatermark was computed on the fly not what user specifies
        partitions.add(new Partition(entry.getKey(), partitionHighWatermark, false));
      }
    }
    return partitions;
  }
949745
def pause_topic(self, topic):
        
        nsq.assert_valid_topic_name(topic)
        return self._request('POST', '/topic/pause', fields={'topic': topic})
638067
private Controllable createTopicMBean(Controllable c) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "createPublicationPointMBean", new Object[] { c });
        TopicImpl pp = new TopicImpl(_me, c);
        controllableMap.put(c, pp);
        try {
            Dictionary<String, Object> properties = new Hashtable<String, Object>();
            properties.put("service.vendor", "IBM");
            String cName = c.getName();
            properties.put("jmx.objectname", "WebSphere:feature=wasJmsServer,type=Topic,name=" + cName);
            ServiceRegistration<TopicMBean> topicMBean = (ServiceRegistration<TopicMBean>) bcontext.registerService(TopicMBean.class.getName(), pp, properties);
            serviceObjects.put(cName, topicMBean);
        } catch (Exception e) {
            SibTr.exception(tc, e);
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "createPublicationPointMBean", new Object[] { pp });
        return pp;

    }
652228
def example_create_partitions(a, topics):
    

    new_parts = [NewPartitions(topic, int(new_total_count)) for
                 topic, new_total_count in zip(topics[0::2], topics[1::2])]

    # Try switching validate_only to True to only validate the operation
    # on the broker but not actually perform it.
    fs = a.create_partitions(new_parts, validate_only=False)

    # Wait for operation to finish.
    for topic, f in fs.items():
        try:
            f.result()  # The result itself is None
            print("Additional partitions created for topic {}".format(topic))
        except Exception as e:
            print("Failed to add partitions to topic {}: {}".format(topic, e))
412701
def create_topic(name, partitions: nil, replication_factor: nil, config: nil)
      raise ArgumentError, "partitions must be a positive integer" if Integer(partitions) <= 0
      raise ArgumentError, "replication_factor must be a positive integer" if Integer(replication_factor) <= 0

      Kazoo::Topic.create(self, name, partitions: Integer(partitions), replication_factor: Integer(replication_factor), config: config)
    end
649317
@Override
	public TopicPublisher createPublisher(Topic topic) throws JMSException
    {
    	externalAccessLock.readLock().lock();
    	try
		{
	        checkNotClosed();
	        LocalTopicPublisher publisher = new LocalTopicPublisher(this,topic,idProvider.createID());
	        registerProducer(publisher);
	        return publisher;
		}
    	finally
    	{
    		externalAccessLock.readLock().unlock();
    	}
    }
163329
private void calculateTrackers(Collection<TopicPartition> partitions) {

        Map<TopicPartition, CommitTracker> trackers = new HashMap<>();
        trackers.putAll(m_trackerMap.get());

        Map<TopicPartition, AtomicLong> lastCommittedOffSets = new HashMap<>();
        lastCommittedOffSets.putAll(m_lastCommittedOffSets.get());

        boolean newTopicPartition = false;
        for (TopicPartition partition : partitions) {
            if (m_trackerMap.get().get(partition) != null) {
                continue;
            }
            newTopicPartition = true;
            long startOffset = -1L;
            CommitTracker commitTracker = null;
            if (m_config.getCommitPolicy() == KafkaCommitPolicy.TIME && m_config.getTriggerValue() > 0) {
                commitTracker = new SimpleTracker();
            } else {
                commitTracker = new DurableTracker(KafkaConstants.IMPORT_GAP_LEAD, partition.topic(),
                        partition.partition(), m_config.getGroupId());
            }
            trackers.put(partition, commitTracker);
            try {
                OffsetAndMetadata offsetAndMetaData = m_consumer.committed(partition);
                startOffset = offsetAndMetaData != null ? offsetAndMetaData.offset() : -1L;
                if (startOffset > -1L) {
                    commitTracker.resetTo(startOffset);
                }
            } catch (KafkaException e) {
                LOGGER.error("Failed to read committed offsets for group " + m_config.getGroupId() + partition + " " + e.getMessage());
            }
            lastCommittedOffSets.put(partition, new AtomicLong(startOffset));
            m_pauseOffsets.put(partition, new AtomicLong(-1));
            m_workTrackers.put(partition, new PendingWorkTracker());
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Starting offset for group:" + m_config.getGroupId() + ":" + startOffset + " partition:" + partition);
            }
        }
        if (newTopicPartition) {
            m_trackerMap.set(trackers);
            m_lastCommittedOffSets.set(lastCommittedOffSets);
        }
    }
957765
def seek(self, offset, topic_partition=None):
        

        if isinstance(topic_partition, TopicPartition):
            self.consumer.seek(topic_partition, offset=offset)
        else:
            raise TypeError("topic_partition must be of type TopicPartition, create it with Create TopicPartition keyword.")
689949
def _build_partitions(self, assignment):
        
        self.partitions = {}
        for partition_name, replica_ids in six.iteritems(assignment):
            # Get topic
            topic_id = partition_name[0]
            partition_id = partition_name[1]
            topic = self.topics.setdefault(
                topic_id,
                Topic(topic_id, replication_factor=len(replica_ids))
            )

            # Creating partition object
            partition = Partition(
                topic,
                partition_id,
                weight=self.partition_measurer.get_weight(partition_name),
                size=self.partition_measurer.get_size(partition_name),
            )
            self.partitions[partition_name] = partition
            topic.add_partition(partition)

            # Updating corresponding broker objects
            for broker_id in replica_ids:
                # Check if broker-id is present in current active brokers
                if broker_id not in list(self.brokers.keys()):
                    self.log.warning(
                        "Broker %s containing partition %s is not in "
                        "active brokers.",
                        broker_id,
                        partition,
                    )
                    self.brokers[broker_id] = self._create_broker(broker_id)

                self.brokers[broker_id].add_partition(partition)
509945
public static void markPartitionAsUnindexed(int partitionId, InternalIndex[] indexes) {
        for (InternalIndex index : indexes) {
            index.markPartitionAsUnindexed(partitionId);
        }
    }
1161708
public static void setPartitionAvgRecordSize(State state, KafkaPartition partition, long size) {
    state.setProp(getPartitionPropName(partition.getTopicName(), partition.getId()) + "." + KafkaSource.AVG_RECORD_SIZE,
        size);
  }