You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2020/05/06 03:53:29 UTC
[GitHub] [druid] jon-wei commented on a change in pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
jon-wei commented on a change in pull request #9819:
URL: https://github.com/apache/druid/pull/9819#discussion_r420526047
##########
File path: extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -679,87 +694,204 @@ public String getPosition(StreamPartition<String> partition)
);
}
- @Override
- public void close()
+ /**
+ * Fetch the partition lag, given a stream and set of current partition offsets. This operates independently from
+ * the {@link PartitionResource} which have been assigned to this record supplier.
+ */
+ public Map<String, Long> getPartitionsTimeLag(String stream, Map<String, String> currentOffsets)
{
- if (this.closed) {
- return;
- }
-
- assign(ImmutableSet.of());
-
- scheduledExec.shutdown();
-
- try {
- if (!scheduledExec.awaitTermination(EXCEPTION_RETRY_DELAY_MS, TimeUnit.MILLISECONDS)) {
- scheduledExec.shutdownNow();
- }
- }
- catch (InterruptedException e) {
- log.warn(e, "InterruptedException while shutting down");
- throw new RuntimeException(e);
+ Map<String, Long> partitionLag = Maps.newHashMapWithExpectedSize(currentOffsets.size());
+ for (Map.Entry<String, String> partitionOffset : currentOffsets.entrySet()) {
+ StreamPartition<String> partition = new StreamPartition<>(stream, partitionOffset.getKey());
+ long currentLag = getPartitionResourcesTimeLag(partition, partitionOffset.getValue());
+ partitionLag.put(partitionOffset.getKey(), currentLag);
}
-
- this.closed = true;
+ return partitionLag;
}
- // this is only used for tests
+ /**
+ * This method is only used for tests to verify that {@link PartitionResource} in fact tracks it's current lag
+ * as it is polled for records. This isn't currently used in production at all, but could be one day if we were
+ * to prefer to get the lag from the running tasks in the same API call which fetches the current task offsets,
+ * instead of directly calling the AWS Kinesis API with the offsets returned from those tasks
+ * (see {@link #getPartitionsTimeLag}, which accepts a map of current partition offsets).
+ */
@VisibleForTesting
- Map<String, Long> getPartitionTimeLag()
+ Map<String, Long> getPartitionResourcesTimeLag()
{
return partitionResources.entrySet()
.stream()
.collect(
- Collectors.toMap(k -> k.getKey().getPartitionId(), k -> k.getValue().getPartitionTimeLag())
+ Collectors.toMap(
+ k -> k.getKey().getPartitionId(),
+ k -> k.getValue().getPartitionTimeLag()
+ )
);
}
- public Map<String, Long> getPartitionTimeLag(Map<String, String> currentOffsets)
- throws InterruptedException, TimeoutException, ExecutionException
+ @VisibleForTesting
+ public int bufferSize()
{
- Map<String, Long> partitionLag = Maps.newHashMapWithExpectedSize(currentOffsets.size());
-
- List<Future<AbstractMap.SimpleEntry<String, Long>>> longo = scheduledExec.invokeAll(
- partitionResources.entrySet()
- .stream()
- .map(partition -> {
- final String partitionId = partition.getKey().getPartitionId();
- return partition.getValue().getLagCallable(partitionId, currentOffsets.get(partitionId));
- })
- .collect(Collectors.toList())
-
- );
- for (Future<AbstractMap.SimpleEntry<String, Long>> future : longo) {
- AbstractMap.SimpleEntry<String, Long> result = future.get(fetchSequenceNumberTimeout, TimeUnit.MILLISECONDS);
- partitionLag.put(result.getKey(), result.getValue());
- }
+ return records.size();
+ }
- return partitionLag;
+ @VisibleForTesting
+ public boolean isBackgroundFetchRunning()
+ {
+ return partitionsFetchStarted.get();
}
- private void seekInternal(StreamPartition<String> partition, String sequenceNumber, ShardIteratorType iteratorEnum)
+ /**
+ * Check that a {@link PartitionResource} has been assigned to this record supplier, and if so call
+ * {@link PartitionResource#seek} to move it to the latest offsets. Note that this method does not restart background
+ * fetch, which should have been stopped prior to calling this method by a call to
+ * {@link #filterBufferAndResetBackgroundFetch}.
+ */
+ private void partitionSeek(StreamPartition<String> partition, String sequenceNumber, ShardIteratorType iteratorEnum)
{
PartitionResource resource = partitionResources.get(partition);
if (resource == null) {
throw new ISE("Partition [%s] has not been assigned", partition);
}
+ resource.seek(iteratorEnum, sequenceNumber);
+ }
- log.debug(
- "Seeking partition [%s] to [%s]",
- partition.getPartitionId(),
- sequenceNumber != null ? sequenceNumber : iteratorEnum.toString()
- );
+ /**
+ * Given a partition and a {@link ShardIteratorType}, create a shard iterator and fetch
+ * {@link #FETCH_SEQUENCE_NUMBER_RECORD_COUNT} records and return the first sequence number from the result set.
+ * This method is thread safe as it does not depend on the internal state of the supplier (it doesn't use the
+ * {@link PartitionResource} which have been assigned to the supplier), and the Kinesis client is thread safe.
+ */
+ @Nullable
+ private String getSequenceNumber(StreamPartition<String> partition, ShardIteratorType iteratorEnum)
+ {
+ return wrapExceptions(() -> {
+ String shardIterator =
+ kinesis.getShardIterator(partition.getStream(), partition.getPartitionId(), iteratorEnum.toString())
+ .getShardIterator();
+ long timeoutMillis = System.currentTimeMillis() + fetchSequenceNumberTimeout;
+ GetRecordsResult recordsResult = null;
+
+ while (shardIterator != null && System.currentTimeMillis() < timeoutMillis) {
+
+ if (closed) {
+ log.info("KinesisRecordSupplier closed while fetching sequenceNumber");
+ return null;
+ }
+ try {
+ // we call getRecords with limit 1000 to make sure that we can find the first (earliest) record in the shard.
+ // In the case where the shard is constantly removing records that are past their retention period, it is possible
+ // that we never find the first record in the shard if we use a limit of 1.
+ recordsResult = kinesis.getRecords(
+ new GetRecordsRequest().withShardIterator(shardIterator).withLimit(FETCH_SEQUENCE_NUMBER_RECORD_COUNT)
+ );
+ }
+ catch (ProvisionedThroughputExceededException e) {
+ log.warn(
+ e,
+ "encountered ProvisionedThroughputExceededException while fetching records, this means "
+ + "that the request rate for the stream is too high, or the requested data is too large for "
+ + "the available throughput. Reduce the frequency or size of your requests. Consider increasing "
+ + "the number of shards to increase throughput."
+ );
+ try {
+ Thread.sleep(PROVISIONED_THROUGHPUT_EXCEEDED_BACKOFF_MS);
+ continue;
+ }
+ catch (InterruptedException e1) {
+ log.warn(e1, "Thread interrupted!");
+ Thread.currentThread().interrupt();
+ break;
+ }
+ }
- resource.shardIterator = wrapExceptions(() -> kinesis.getShardIterator(
- partition.getStream(),
- partition.getPartitionId(),
- iteratorEnum.toString(),
- sequenceNumber
- ).getShardIterator());
+ List<Record> records = recordsResult.getRecords();
+
+ if (!records.isEmpty()) {
+ return records.get(0).getSequenceNumber();
+ }
+
+ shardIterator = recordsResult.getNextShardIterator();
+ }
+
+ if (shardIterator == null) {
+ log.info("Partition[%s] returned a null shard iterator, is the shard closed?", partition.getPartitionId());
+ return KinesisSequenceNumber.END_OF_SHARD_MARKER;
+ }
+
+
+ // if we reach here, it usually means either the shard has no more records, or records have not been
+ // added to this shard
+ log.warn(
+ "timed out while trying to fetch position for shard[%s], millisBehindLatest is [%s], likely no more records in shard",
+ partition.getPartitionId(),
+ recordsResult != null ? recordsResult.getMillisBehindLatest() : "UNKNOWN"
+ );
+ return null;
+ });
}
- private void filterBufferAndResetFetchRunnable(Set<StreamPartition<String>> partitions) throws InterruptedException
+ /**
+ * Given a {@link StreamPartition} and an offset, create a 'shard iterator' for the offset and fetch a single record
+ * in order to get the lag: {@link GetRecordsResult#getMillisBehindLatest()}. This method is thread safe as it does
+ * not depend on the internal state of the supplier (it doesn't use the {@link PartitionResource} which have been
+ * assigned to the supplier), and the Kinesis client is thread safe.
+ */
+ private Long getPartitionResourcesTimeLag(StreamPartition<String> partition, String offset)
Review comment:
Suggest renaming this to not have `Resources` in the name since it doesn't use PartitionResource
##########
File path: core/src/main/java/org/apache/druid/indexer/TaskIdUtils.java
##########
@@ -31,6 +32,8 @@
{
private static final Pattern INVALIDCHARS = Pattern.compile("(?s).*[^\\S ].*");
+ private static final Joiner underscoreJoiner = Joiner.on("_");
Review comment:
I think `underscoreJoiner` should be all caps
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org