create logs with given partition number

@param topic        the topic name
@param partitions   partition number
@param forceEnlarge enlarge the partition number of log if smaller than runtime
@return the partition number of the log after enlarging
552360
public int createLogs(String topic, final int partitions, final boolean forceEnlarge) {
        TopicNameValidator.validate(topic);
        synchronized (logCreationLock) {
            final int configPartitions = getPartition(topic);
            if (configPartitions >= partitions || !forceEnlarge) {
                return configPartitions;
            }
            topicPartitionsMap.put(topic, partitions);
            if (config.getEnableZookeeper()) {
                if (getLogPool(topic, 0) != null) {//created already
                    topicRegisterTasks.add(new TopicTask(TopicTask.TaskType.ENLARGE, topic));
                } else {
                    topicRegisterTasks.add(new TopicTask(TopicTask.TaskType.CREATE, topic));
                }
            }
            return partitions;
        }
    }
854185
function Logsene (token, type, url, storageDirectory, options) {
  if (!token) {
    throw new Error('Logsene token not specified')
  }
  if (options) {
    this.options = options
  } else {
    this.options = {
      useIndexInBulkUrl: false
    }
  }
  if (url && /logsene/.test(url)) {
    // logs to logsene should use /TOKEN/_bulk
    this.options.useIndexInBulkUrl = true
  }
  this.request = null
  this.maxMessageFieldSize = MAX_MESSAGE_FIELD_SIZE
  this.xLogseneOrigin = xLogseneOrigin
  this.token = token
  this.setUrl(url || process.env.LOGSENE_URL || process.env.LOGSENE_RECEIVER_URL || process.env.LOGS_URL || 'https://logsene-receiver.sematext.com/_bulk')
  this.type = type || 'logs'
  this.hostname = process.env.SPM_REPORTED_HOSTNAME || os.hostname()
  this.bulkReq = new streamBuffers.WritableStreamBuffer({
    initialSize: initialBufferSize,
    incrementAmount: incrementBuffer
  })
  this.logCount = 0
  this.sourceName = null
  if (process.mainModule && process.mainModule.filename) {
    this.sourceName = path.basename(process.mainModule.filename)
  }
  events.EventEmitter.call(this)
  var self = this
  self.lastSend = Date.now()
  var logInterval = Number(process.env.LOGSENE_LOG_INTERVAL) || 20000
  var tid = setInterval(function () {
    if (self.logCount > 0 && (Date.now() - self.lastSend) > (logInterval - 1000)) {
      self.send()
    }
  }, logInterval)
  if (tid.unref) {
    tid.unref()
  }
  process.on('beforeExit', function () {
    self.send()
  })
  if (process.env.LOGSENE_TMP_DIR || storageDirectory) {
    this.diskBuffer(true, process.env.LOGSENE_TMP_DIR || storageDirectory)
  }
  var fieldListStr = process.env.LOGSENE_REMOVE_FIELDS || process.env.REMOVE_FIELDS || ''
  this.removeFieldsList = fieldListStr.replace(/ /g, '').split(',')
}
1171909
protected static long calculateFileSplit(long repositorySize) {
		if (repositorySize <= 0) {
			return MAX_LOG_FILE_SIZE;
		}
		if (repositorySize < MIN_REPOSITORY_SIZE) {
			throw new IllegalArgumentException("Specified repository size is too small");
		}
		
		long result = repositorySize / SPLIT_RATIO;
		
		if (result < MIN_LOG_FILE_SIZE) {
			result = MIN_LOG_FILE_SIZE;
		} else if (result > MAX_LOG_FILE_SIZE) {
			result = MAX_LOG_FILE_SIZE;
		}
		
		return result;
	}
903954
public ProxyDataSourceBuilder logSlowQueryByCommons(long thresholdTime, TimeUnit timeUnit, CommonsLogLevel logLevel) {
        return logSlowQueryByCommons(thresholdTime, timeUnit, logLevel, null);
    }
943646
@Override
	public void createPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, CatalogPartition partition, boolean ignoreIfExists)
			throws TableNotExistException, TableNotPartitionedException, PartitionSpecInvalidException, PartitionAlreadyExistsException, CatalogException {
		throw new UnsupportedOperationException();
	}
529598
public ProxyDataSourceBuilder logSlowQueryByCommons(long thresholdTime, TimeUnit timeUnit, String logName) {
        return logSlowQueryByCommons(thresholdTime, timeUnit, null, logName);
    }
861816
private static List<Integer> repartitionStreamlet(int incomingInteger, int numPartitions) {
    List<Integer> partitions;

    if (incomingInteger >= 0 && incomingInteger < 25) {
      partitions = Arrays.asList(0, 1);
    } else if (incomingInteger > 26 && incomingInteger < 50) {
      partitions = Arrays.asList(2, 3);
    } else if (incomingInteger > 50 && incomingInteger < 75) {
      partitions = Arrays.asList(4, 5);
    } else if (incomingInteger > 76 && incomingInteger <= 100) {
      partitions = Arrays.asList(6, 7);
    } else {
      partitions = Arrays.asList(ThreadLocalRandom.current().nextInt(0, 8));
    }

    String logMessage = String.format("Sending value %d to partitions: %s",
        incomingInteger,
        StreamletUtils.intListAsString(partitions));

    LOG.info(logMessage);

    return partitions;
  }
1062146
protected void setLogFileSize(long newFileSize)
                    throws ObjectManagerException
    {
        if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
            trace.entry(this, cclass
                        , "setLogFileSize"
                        , "newFileSize=" + newFileSize + "(long)"
                            );

        // Note where we are in the log file.
        long initialFilePosition = filePosition;

        synchronized (newFileSizeRequestLock) {
            // Force the new file size to be a whole number of pages.
            this.newFileSize = (newFileSize / pageSize) * pageSize;
            newFileSizeRequested = true;
            newFileSizeException = null;
            logFileSizeRequested = newFileSize;
            // We have not yet seen the log file wrap.
            boolean wrapped = false;

            // Keep requesting checkpoints until the start and end of the log data are both 
            // before the new end of the log, without the user log data wrapping round the 
            // end of the logFile.
            while (newFileSizeRequested) {

                // Check there is still some chance of fitting the current log into the new one.
                // and that we have not wrapped round the log file without succeeding.
                long fileSpaceAvailable = getLogFileSpaceLeft();
                long newFileSpaceAvailable = fileSpaceAvailable - fileLogHeader.fileSize + newFileSize;
                float newOcupancy = (float) 1.0 - (float) (newFileSpaceAvailable) / (float) (newFileSize);
                if (newOcupancy > objectManagerState.logFullPostCheckpointThreshold
                    || wrapped) {
                    newFileSizeRequested = false;
                    logFileSizeRequested = 0;
                    if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
                        trace.exit(this,
                                   cclass,
                                   "setLogFileSize",
                                   new Object[] { new Long(fileLogHeader.fileSize),
                                                 new Long(newFileSize),
                                                 new Float(newOcupancy) });
                    throw new LogFileSizeTooSmallException(this
                                                           , fileLogHeader.fileSize
                                                           , newFileSize
                                                           , fileSpaceAvailable
                                                           , newOcupancy
                                                           , objectManagerState.logFullPostCheckpointThreshold);
                } // if (   (fileSpaceLeft...

                if (Tracing.isAnyTracingEnabled() && trace.isDebugEnabled())
                    trace.debug(this,
                                cclass,
                                "setLogFileSize",
                                new Object[] { new Long(filePosition),
                                              new Long(initialFilePosition) });

                // If we have wrapped round the log file, give up after the next attempt to satisfy the 
                // request. 
                if (filePosition < initialFilePosition)
                    wrapped = true;

                // We have released the lock on writingFlushSet set so now we can request a checkpoint and
                // see if we are successful.
                objectManagerState.waitForCheckpoint(true);

                // Speed up moving the filePosition by writing a reasonable ammount of data.
                // The folowing is dangerous because it can cause the request to fail if we 
                // are trying to reduce the log file size close to the minimum possible. 
//        PaddingLogRecord paddingLogRecord = new PaddingLogRecord(logBufferSize/2);
//        writeNext(paddingLogRecord
//                 ,0
//                 ,true
//                 );

            } // while (newFileSizeRequested...

            logFileSizeRequested = 0;
        } // synchronized (newFileSizeRequestLock).                                            

        if (newFileSizeException != null) {
            if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
                trace.exit(this, cclass
                           , "setLogFileSize"
                           , "via PermanentIOException newFileSizeException=" + newFileSizeException + "(PermanentIOException)"
                                );
            throw newFileSizeException;
        }

        if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
            trace.exit(this, cclass
                       , "setLogFileSize"
                            );
    }
179798
private Partition getPartitionWithCreateTime(Partition partition, int createTime) {
    if (partition.isSetCreateTime() && partition.getCreateTime() > 0) {
      return partition;
    }
    Partition actualPartition = partition.deepCopy();
    actualPartition.setCreateTime(createTime);
    return actualPartition;
  }
958385
public void setClusterLogging(java.util.Collection<LogSetup> clusterLogging) {
        if (clusterLogging == null) {
            this.clusterLogging = null;
            return;
        }

        this.clusterLogging = new java.util.ArrayList<LogSetup>(clusterLogging);
    }
257754
public static ResultPartitionDeploymentDescriptor from(
			IntermediateResultPartition partition, int maxParallelism, boolean lazyScheduling) {

		final IntermediateDataSetID resultId = partition.getIntermediateResult().getId();
		final IntermediateResultPartitionID partitionId = partition.getPartitionId();
		final ResultPartitionType partitionType = partition.getIntermediateResult().getResultType();

		// The produced data is partitioned among a number of subpartitions.
		//
		// If no consumers are known at this point, we use a single subpartition, otherwise we have
		// one for each consuming sub task.
		int numberOfSubpartitions = 1;

		if (!partition.getConsumers().isEmpty() && !partition.getConsumers().get(0).isEmpty()) {

			if (partition.getConsumers().size() > 1) {
				throw new IllegalStateException("Currently, only a single consumer group per partition is supported.");
			}

			numberOfSubpartitions = partition.getConsumers().get(0).size();
		}

		return new ResultPartitionDeploymentDescriptor(
				resultId, partitionId, partitionType, numberOfSubpartitions, maxParallelism, lazyScheduling);
	}
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);
  }
301896
private void createPartitions() {

        for (Map.Entry<String, String> partitionEntry : this.partitions.entrySet()) {
            try {
                this.addPartitionInternal(partitionEntry.getKey(), partitionEntry.getValue());
            } catch (Exception e) { //NOSONAR
                throw new AssertionError("Could not create partitions " + this.partitions, e);
            }
        }
    }
1185991
public Partition addPartition(String partitionId, String partitionDn) throws Exception {
        Log.info(c, "addPartition", "entry " + partitionId + " " + partitionDn);
        JdbmPartition partition = new JdbmPartition(this.service.getSchemaManager());
        partition.setId(partitionId);
        partition.setPartitionPath(new File(this.service.getInstanceLayout().getPartitionsDirectory(), partitionId).toURI());
        partition.setSuffixDn(new Dn(partitionDn));
        service.addPartition(partition);

        Log.info(c, "addPartition", "exit");
        return partition;
    }
675401
public static long getPartitionAvgRecordSize(State state, KafkaPartition partition) {
    return state.getPropAsLong(
        getPartitionPropName(partition.getTopicName(), partition.getId()) + "." + KafkaSource.AVG_RECORD_SIZE);
  }
1172999
@Override
    public Partition[] getPartitions(ExtractorConfig<T> config) {

        GetPartitionsAction<T> getPartitionsAction = new GetPartitionsAction<>(config);

        channel.writeAndFlush(getPartitionsAction);

        Response response;
        boolean interrupted = false;
        for (; ; ) {
            try {
                response = answer.take();
                break;
            } catch (InterruptedException ignore) {
                interrupted = true;
            }
        }

        if (interrupted) {
            Thread.currentThread().interrupt();
        }

        return ((GetPartitionsResponse) response).getPartitions();
    }
317208
static List<PartitionDefinition> findSlicePartitions(RestClient client, Settings settings, MappingSet mappingSet,
                                                         Map<String, NodeInfo> nodes, List<List<Map<String, Object>>> shards, Log log) {
        QueryBuilder query = QueryUtils.parseQueryAndFilters(settings);
        Integer maxDocsPerPartition = settings.getMaxDocsPerPartition();
        Assert.notNull(maxDocsPerPartition, "Attempting to find slice partitions but maximum documents per partition is not set.");
        Resource readResource = new Resource(settings, true);
        Mapping resolvedMapping = mappingSet == null ? null : mappingSet.getResolvedView();

        List<PartitionDefinition> partitions = new ArrayList<PartitionDefinition>(shards.size());
        for (List<Map<String, Object>> group : shards) {
            String index = null;
            int shardId = -1;
            List<String> locationList = new ArrayList<String> ();
            for (Map<String, Object> replica : group) {
                ShardInfo shard = new ShardInfo(replica);
                index = shard.getIndex();
                shardId = shard.getName();
                if (nodes.containsKey(shard.getNode())) {
                    locationList.add(nodes.get(shard.getNode()).getPublishAddress());
                }
            }
            String[] locations = locationList.toArray(new String[0]);
            if (index == null) {
                // Could not find shards for this partition. Continue anyway?
                if (settings.getIndexReadAllowRedStatus()) {
                    log.warn("Shard information is missing from an index and will not be reached during job execution. " +
                            "Assuming shard is unavailable and cluster is red! Continuing with read operation by " +
                            "skipping this shard! This may result in incomplete data retrieval!");
                } else {
                    throw new IllegalStateException("Could not locate shard information for one of the read indices. " +
                            "Check your cluster status to see if it is unstable!");
                }
            } else {
                // TODO applyAliasMetaData should be called in order to ensure that the count are exact (alias filters and routing may change the number of documents)
                long numDocs;
                if (readResource.isTyped()) {
                    numDocs = client.count(index, readResource.type(), Integer.toString(shardId), query);
                } else {
                    numDocs = client.countIndexShard(index, Integer.toString(shardId), query);
                }
                int numPartitions = (int) Math.max(1, numDocs / maxDocsPerPartition);
                for (int i = 0; i < numPartitions; i++) {
                    PartitionDefinition.Slice slice = new PartitionDefinition.Slice(i, numPartitions);
                    partitions.add(new PartitionDefinition(settings, resolvedMapping, index, shardId, slice, locations));
                }
            }
        }
        return partitions;
    }