Shortcut variant of {@link #createPartitionStateHolders(Map, int, SerializedValue, SerializedValue, ClassLoader)}
that uses the same offset for all partitions when creating their state holders.
546676
private List<KafkaTopicPartitionState<KPH>> createPartitionStateHolders(
		List<KafkaTopicPartition> partitions,
		long initialOffset,
		int timestampWatermarkMode,
		SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic,
		SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated,
		ClassLoader userCodeClassLoader) throws IOException, ClassNotFoundException {

		Map<KafkaTopicPartition, Long> partitionsToInitialOffset = new HashMap<>(partitions.size());
		for (KafkaTopicPartition partition : partitions) {
			partitionsToInitialOffset.put(partition, initialOffset);
		}

		return createPartitionStateHolders(
				partitionsToInitialOffset,
				timestampWatermarkMode,
				watermarksPeriodic,
				watermarksPunctuated,
				userCodeClassLoader);
	}
433574
@Override
    public void setPartitions(Partitions partitions) {

        boolean reconfigurePartitions = false;

        synchronized (stateLock) {
            if (this.partitions != null) {
                reconfigurePartitions = true;
            }
            this.partitions = partitions;
            this.connectionFactory.setPartitions(partitions);
        }

        if (reconfigurePartitions) {
            reconfigurePartitions();
        }
    }
749087
private DBObject createQueryPartition(MongoPartition partition) {

        QueryBuilder queryBuilderMin = QueryBuilder.start(partition.getKey());
        DBObject bsonObjectMin = queryBuilderMin.greaterThanEquals(partition.splitWrapper().getStartToken()).get();

        QueryBuilder queryBuilderMax = QueryBuilder.start(partition.getKey());
        DBObject bsonObjectMax = queryBuilderMax.lessThan(partition.splitWrapper().getEndToken()).get();

        QueryBuilder queryBuilder = QueryBuilder.start();
        if (partition.splitWrapper().getStartToken() != null) {
            queryBuilder.and(bsonObjectMin);
        }

        if (partition.splitWrapper().getEndToken() != null) {
            queryBuilder.and(bsonObjectMax);
        }

        LOG.debug("mongodb query "+queryBuilder.get());

        return queryBuilder.get();
    }
229272
@SuppressWarnings("unchecked")
  void maybeMovePartition(final short partition) {
    Observable
        .timer(50, TimeUnit.MILLISECONDS)
        .filter(ignored -> {
          PartitionState ps = sessionState.get(partition);
          boolean desiredSeqnoReached = ps.isAtEnd();
          if (desiredSeqnoReached) {
            LOGGER.debug("Reached desired high seqno {} for vbucket {}, not reopening stream.",
                ps.getEndSeqno(), partition);
          }
          return !desiredSeqnoReached;
        })
        .flatMapCompletable(ignored -> {
          PartitionState ps = sessionState.get(partition);
          return startStreamForPartition(
              partition,
              ps.getLastUuid(),
              ps.getStartSeqno(),
              ps.getEndSeqno(),
              ps.getSnapshotStartSeqno(),
              ps.getSnapshotEndSeqno()
          ).retryWhen(anyOf(NotMyVbucketException.class)
              .max(Integer.MAX_VALUE)
              .delay(Delay.fixed(200, TimeUnit.MILLISECONDS))
              .build());
        }).toCompletable().subscribe(new CompletableSubscriber() {
      @Override
      public void onCompleted() {
        LOGGER.trace("Completed Partition Move for partition {}", partition);
      }

      @Override
      public void onError(Throwable e) {
        if (e instanceof RollbackException) {
          // Benign, so don't log a scary stack trace. A synthetic "rollback" message has been passed
          // to the Control Event Handler, which may react by calling Client.rollbackAndRestartStream().
          LOGGER.warn("Rollback during Partition Move for partition {}", partition);
        } else {
          LOGGER.warn("Error during Partition Move for partition {}", partition, e);
        }
        if (env.eventBus() != null) {
          env.eventBus().publish(new FailedToMovePartitionEvent(partition, e));
        }
      }

      @Override
      public void onSubscribe(Subscription d) {
        LOGGER.debug("Subscribing for Partition Move for partition {}", partition);
      }
    });
  }
894247
private List<PartETag> collectPartETags() {

		final List<PartETag> partETags = new ArrayList<PartETag>();
		for (Future<PartETag> future : futures) {
			try {
				partETags.add(future.get());
			} catch (Exception e) {
				throw new SdkClientException("Unable to copy part: "
						+ e.getCause().getMessage(), e.getCause());
			}
		}
		return partETags;
	}
658999
@Override
	public TopicPartition createKafkaPartitionHandle(KafkaTopicPartition partition) {
		return new TopicPartition(partition.getTopic(), partition.getPartition());
	}
98459
public Partition withParameters(java.util.Map<String, String> parameters) {
        setParameters(parameters);
        return this;
    }
630366
@Override
	protected TopicAndPartition createKafkaPartitionHandle(KafkaTopicPartition partition) {
		return new TopicAndPartition(partition.getTopic(), partition.getPartition());
	}
88055
public Collection<Map.Entry<Address, List<Integer>>> getPartitions(PartitionTableView partitionTableView) {

        Map<Address, List<Integer>> partitionsMap = new HashMap<Address, List<Integer>>();

        int partitionCount = partitionTableView.getLength();
        for (int partitionId = 0; partitionId < partitionCount; partitionId++) {
            PartitionReplica owner = partitionTableView.getReplica(partitionId, 0);
            if (owner == null) {
                partitionsMap.clear();
                return partitionsMap.entrySet();
            }
            Address clientOwnerAddress = clientAddressOf(owner.address());
            if (clientOwnerAddress == null) {
                partitionsMap.clear();
                return partitionsMap.entrySet();
            }
            List<Integer> indexes = partitionsMap.get(clientOwnerAddress);
            if (indexes == null) {
                indexes = new LinkedList<Integer>();
                partitionsMap.put(clientOwnerAddress, indexes);
            }
            indexes.add(partitionId);
        }
        return partitionsMap.entrySet();
    }
784729
private <K> DataStream<T> partitionCustom(Partitioner<K> partitioner, Keys<T> keys) {
		KeySelector<T, K> keySelector = KeySelectorUtil.getSelectorForOneKey(keys, partitioner, getType(), getExecutionConfig());

		return setConnectionType(
				new CustomPartitionerWrapper<>(
						clean(partitioner),
						clean(keySelector)));
	}
737466
public static HivePartition getHivePartition(Partition partition) {
    State partitionProps = getPartitionProps(partition);
    State storageProps = getStorageProps(partition.getSd());
    State serDeProps = getSerDeProps(partition.getSd().getSerdeInfo());
    HivePartition hivePartition =
        new HivePartition.Builder().withDbName(partition.getDbName()).withTableName(partition.getTableName())
            .withPartitionValues(partition.getValues()).withProps(partitionProps).withStorageProps(storageProps)
            .withSerdeProps(serDeProps).build();
    if (partition.getCreateTime() > 0) {
      hivePartition.setCreateTime(partition.getCreateTime());
    }
    if (partition.getSd().getCols() != null) {
      hivePartition.setColumns(getColumns(partition.getSd().getCols()));
    }
    if (partition.getSd().getBucketCols() != null) {
      hivePartition.setBucketColumns(partition.getSd().getBucketCols());
    }
    return hivePartition;
  }
480149
public HashMap<KafkaTopicPartition, Long> snapshotCurrentState() {
		// this method assumes that the checkpoint lock is held
		assert Thread.holdsLock(checkpointLock);

		HashMap<KafkaTopicPartition, Long> state = new HashMap<>(subscribedPartitionStates.size());
		for (KafkaTopicPartitionState<KPH> partition : subscribedPartitionStates) {
			state.put(partition.getKafkaTopicPartition(), partition.getOffset());
		}
		return state;
	}
698777
public boolean setPartitions(Collection<Collection<Node>> parts) {
        if (!isDisjoint(parts)) {
            return false;
        }
        partitions = parts;
        return true;
    }
964376
@Override
    public CompletableFuture<Void> createAllLeasesIfNotExists(List<String> partitionIds) {
        CompletableFuture<Void> future = null;

        // Optimization: list the blobs currently existing in the directory. If there are the
        // expected number of blobs, then we can skip doing the creates.
        int blobCount = 0;
        try {
            Iterable<ListBlobItem> leaseBlobs = this.consumerGroupDirectory.listBlobs("", true, null, this.leaseOperationOptions, null);
            Iterator<ListBlobItem> blobIterator = leaseBlobs.iterator();
            while (blobIterator.hasNext()) {
                blobCount++;
                blobIterator.next();
            }
        } catch (URISyntaxException | StorageException e) {
            TRACE_LOGGER.error(this.hostContext.withHost("Exception checking lease existence - leaseContainerName: " + this.storageContainerName + " consumerGroupName: "
                    + this.hostContext.getConsumerGroupName() + " storageBlobPrefix: " + this.storageBlobPrefix), e);
            future = new CompletableFuture<Void>();
            future.completeExceptionally(LoggingUtils.wrapException(e, EventProcessorHostActionStrings.CREATING_LEASES));
        }

        if (future == null) {
            // No error checking the list, so keep going
            if (blobCount == partitionIds.size()) {
                // All expected blobs found, so short-circuit
                future = CompletableFuture.completedFuture(null);
            } else {
                // Create the blobs in parallel
                ArrayList<CompletableFuture<CompleteLease>> createFutures = new ArrayList<CompletableFuture<CompleteLease>>();

                for (String id : partitionIds) {
                    CompletableFuture<CompleteLease> oneCreate = CompletableFuture.supplyAsync(() -> {
                        CompleteLease returnLease = null;
                        try {
                            returnLease = createLeaseIfNotExistsInternal(id, this.leaseOperationOptions);
                        } catch (URISyntaxException | IOException | StorageException e) {
                            TRACE_LOGGER.error(this.hostContext.withHostAndPartition(id,
                                    "Exception creating lease - leaseContainerName: " + this.storageContainerName + " consumerGroupName: " + this.hostContext.getConsumerGroupName()
                                            + " storageBlobPrefix: " + this.storageBlobPrefix), e);
                            throw LoggingUtils.wrapException(e, EventProcessorHostActionStrings.CREATING_LEASES);
                        }
                        return returnLease;
                    }, this.hostContext.getExecutor());
                    createFutures.add(oneCreate);
                }

                CompletableFuture<?>[] dummy = new CompletableFuture<?>[createFutures.size()];
                future = CompletableFuture.allOf(createFutures.toArray(dummy));
            }
        }

        return future;
    }
152137
public void setPartitionValues(java.util.Collection<String> partitionValues) {
        if (partitionValues == null) {
            this.partitionValues = null;
            return;
        }

        this.partitionValues = new java.util.ArrayList<String>(partitionValues);
    }
491882
private CacheTopology addPartitioner(CacheTopology cacheTopology) {
      ConsistentHash currentCH = cacheTopology.getCurrentCH();
      currentCH = new PartitionerConsistentHash(currentCH, keyPartitioner);
      ConsistentHash pendingCH = cacheTopology.getPendingCH();
      if (pendingCH != null) {
         pendingCH = new PartitionerConsistentHash(pendingCH, keyPartitioner);
      }
      ConsistentHash unionCH = cacheTopology.getUnionCH();
      if (unionCH != null) {
         unionCH = new PartitionerConsistentHash(unionCH, keyPartitioner);
      }
      return new CacheTopology(cacheTopology.getTopologyId(), cacheTopology.getRebalanceId(), currentCH, pendingCH,
            unionCH, cacheTopology.getPhase(), cacheTopology.getActualMembers(), cacheTopology.getMembersPersistentUUIDs());
   }
196494
private PartitionVertexAggregate<M> getPartitioned(long vertexId) {
        assert idManager.isPartitionedVertex(vertexId);
        vertexId=getCanonicalId(vertexId);
        PartitionVertexAggregate<M> state = partitionVertices.get(vertexId);
        if (state==null) {
            partitionVertices.putIfAbsent(vertexId,new PartitionVertexAggregate<>(previousScopes));
            state = partitionVertices.get(vertexId);
        }
        return state;
    }