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/04 04:42:56 UTC
[GitHub] [druid] clintropolis opened a new pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
clintropolis opened a new pull request #9819:
URL: https://github.com/apache/druid/pull/9819
Fixes #TBD.
### Description
TBD
<hr>
This PR has:
- [ ] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/licenses.yaml)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths.
- [ ] added integration tests.
- [x] been tested in a test Druid cluster.
<hr>
##### Key changed/added classes in this PR
* `SeekableStreamSupervisor`
* `KinesisSupervisor`
* `KinesisRecordSupplier`
* `KafkaSupervisor`
----------------------------------------------------------------
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
[GitHub] [druid] clintropolis commented on a change in pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
Posted by GitBox <gi...@apache.org>.
clintropolis commented on a change in pull request #9819:
URL: https://github.com/apache/druid/pull/9819#discussion_r425674983
##########
File path: extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -665,180 +694,185 @@ 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 = getPartitionTimeLag(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)
+ @VisibleForTesting
+ public int bufferSize()
{
- Map<String, Long> partitionLag = Maps.newHashMapWithExpectedSize(currentOffsets.size());
- for (Map.Entry<StreamPartition<String>, PartitionResource> partition : partitionResources.entrySet()) {
- final String partitionId = partition.getKey().getPartitionId();
- partitionLag.put(partitionId, partition.getValue().getPartitionTimeLag(currentOffsets.get(partitionId)));
- }
- return partitionLag;
+ return records.size();
}
- private void seekInternal(StreamPartition<String> partition, String sequenceNumber, ShardIteratorType iteratorEnum)
+ @VisibleForTesting
+ public boolean isBackgroundFetchRunning()
{
- PartitionResource resource = partitionResources.get(partition);
- if (resource == null) {
- throw new ISE("Partition [%s] has not been assigned", partition);
- }
-
- log.debug(
- "Seeking partition [%s] to [%s]",
- partition.getPartitionId(),
- sequenceNumber != null ? sequenceNumber : iteratorEnum.toString()
- );
-
- resource.shardIterator = wrapExceptions(() -> kinesis.getShardIterator(
- partition.getStream(),
- partition.getPartitionId(),
- iteratorEnum.toString(),
- sequenceNumber
- ).getShardIterator());
-
- checkPartitionsStarted = true;
+ return partitionsFetchStarted.get();
}
- private void filterBufferAndResetFetchRunnable(Set<StreamPartition<String>> partitions) throws InterruptedException
+ /**
+ * 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)
{
- 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 e;
+ PartitionResource resource = partitionResources.get(partition);
+ if (resource == null) {
+ throw new ISE("Partition [%s] has not been assigned", partition);
}
-
- scheduledExec = Executors.newScheduledThreadPool(
- fetchThreads,
- Execs.makeThreadFactory("KinesisRecordSupplier-Worker-%d")
- );
-
- // filter records in buffer and only retain ones whose partition was not seeked
- BlockingQueue<OrderedPartitionableRecord<String, String>> newQ = new LinkedBlockingQueue<>(recordBufferSize);
-
- records.stream()
- .filter(x -> !partitions.contains(x.getStreamPartition()))
- .forEachOrdered(newQ::offer);
-
- records = newQ;
-
- // restart fetching threads
- partitionResources.values().forEach(x -> x.started = false);
- checkPartitionsStarted = true;
- }
-
- @Nullable
- private String getSequenceNumberInternal(StreamPartition<String> partition, ShardIteratorType iteratorEnum)
- {
- return wrapExceptions(() -> getSequenceNumberInternal(
- partition,
- kinesis.getShardIterator(partition.getStream(), partition.getPartitionId(), iteratorEnum.toString())
- .getShardIterator()
- ));
+ resource.seek(iteratorEnum, sequenceNumber);
}
+ /**
+ * 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 getSequenceNumberInternal(StreamPartition<String> partition, String shardIterator)
+ private String getSequenceNumber(StreamPartition<String> partition, ShardIteratorType iteratorEnum)
{
- 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(1000));
- }
- 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."
- );
+ 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 {
- Thread.sleep(PROVISIONED_THROUGHPUT_EXCEEDED_BACKOFF_MS);
- continue;
+ // 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 (InterruptedException e1) {
- log.warn(e1, "Thread interrupted!");
- Thread.currentThread().interrupt();
- break;
+ 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);
Review comment:
It seems better to me, and more consistent with the rest of the codebase, though I'm not sure what to do for retry count since the current behavior would be an indefinite number of retries in the case of `ProvisionedThroughputExceededException`. However, I picked 10, though I can do higher if we are worried this will commonly happen...
##########
File path: extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -112,100 +145,75 @@ private static boolean isServiceExceptionRecoverable(AmazonServiceException ex)
// to indicate that this shard has no more records to read
@Nullable
private volatile String shardIterator;
- private volatile boolean started;
- private volatile boolean stopRequested;
-
private volatile long currentLagMillis;
- PartitionResource(StreamPartition<String> streamPartition)
+ private final AtomicBoolean fetchStarted = new AtomicBoolean();
+
+ private PartitionResource(StreamPartition<String> streamPartition)
{
this.streamPartition = streamPartition;
}
- void startBackgroundFetch()
+ private void startBackgroundFetch()
{
- if (started) {
+ if (!backgroundFetchEnabled) {
return;
}
+ // if seek has been called
+ if (shardIterator == null) {
+ log.warn(
+ "Skipping background fetch for stream[%s] partition[%s] since seek has not been called for this partition",
+ streamPartition.getStream(),
+ streamPartition.getPartitionId()
+ );
+ return;
+ }
+ if (fetchStarted.compareAndSet(false, true)) {
+ log.info(
+ "Starting scheduled fetch for stream[%s] partition[%s]",
+ streamPartition.getStream(),
+ streamPartition.getPartitionId()
+ );
- log.info(
- "Starting scheduled fetch runnable for stream[%s] partition[%s]",
- streamPartition.getStream(),
- streamPartition.getPartitionId()
- );
-
- stopRequested = false;
- started = true;
-
- rescheduleRunnable(fetchDelayMillis);
- }
-
- void stopBackgroundFetch()
- {
- log.info(
- "Stopping scheduled fetch runnable for stream[%s] partition[%s]",
- streamPartition.getStream(),
- streamPartition.getPartitionId()
- );
- stopRequested = true;
+ scheduleBackgroundFetch(fetchDelayMillis);
+ }
}
- long getPartitionTimeLag()
+ private void stopBackgroundFetch()
{
- return currentLagMillis;
+ if (fetchStarted.compareAndSet(true, false)) {
+ log.info(
Review comment:
Modified it to save the `ScheduledFuture` as `currentFetch` and cancel it if it isn't done or null.
----------------------------------------------------------------
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
[GitHub] [druid] clintropolis commented on a change in pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
Posted by GitBox <gi...@apache.org>.
clintropolis commented on a change in pull request #9819:
URL: https://github.com/apache/druid/pull/9819#discussion_r425676717
##########
File path: extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java
##########
@@ -334,16 +334,35 @@ protected boolean useExclusiveStartSequenceNumberForNonFirstSequence()
}
@Override
- protected void updateLatestSequenceFromStream(
- RecordSupplier<Integer, Long> recordSupplier,
- Set<StreamPartition<Integer>> partitions
- )
+ protected void updatePartitionLagFromStream()
Review comment:
We still need to keep `latestSequenceFromStream` around since it's also included in the supervisor report payload, but I guess we could precompute the record lag here as well instead of lazily from the current highest offsets and `latestSequenceFromStream` whenever it is used in a report or emitted as lag metric and it probably wouldn't change the behavior that much.
Though, since generating a report also would be using the highest current offsets and including them in the payload, it might be strange if the record lag didn't match what you could manually compute from the reports current and highest offsets. Because of this report dissonance, I haven't changed it to precompute it yet, but I did leave a comment about what is going on and why we are only fetching the latest offsets instead of doing anything about lag.
----------------------------------------------------------------
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
[GitHub] [druid] jon-wei commented on a change in pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
Posted by GitBox <gi...@apache.org>.
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
[GitHub] [druid] jihoonson commented on pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
Posted by GitBox <gi...@apache.org>.
jihoonson commented on pull request #9819:
URL: https://github.com/apache/druid/pull/9819#issuecomment-628916632
Forgot to mention, probably it would be nice if we split `RecordSupplier` for the supervisor and the task. This could be done as a follow-up.
----------------------------------------------------------------
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
[GitHub] [druid] jihoonson merged pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
Posted by GitBox <gi...@apache.org>.
jihoonson merged pull request #9819:
URL: https://github.com/apache/druid/pull/9819
----------------------------------------------------------------
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
[GitHub] [druid] clintropolis commented on pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
Posted by GitBox <gi...@apache.org>.
clintropolis commented on pull request #9819:
URL: https://github.com/apache/druid/pull/9819#issuecomment-629137464
>Forgot to mention, probably it would be nice if we split RecordSupplier for the supervisor and the task. This could be done as a follow-up.
This seems nice, I considered doing it as part of this PR, but didn't want to create too many moving parts. It is true that the supervisor only needs a subset of the `RecordSupplier` interface and that tasks don't need many of the things the supervisor uses:
* `getPartitionIds`
* `getEarliestSequenceNumber`
* `getLatestSequenceNumber` (only used by supervisor)
* `assign`
* `getAssignment` (only used by the supervisor)
* `seekToEarliest` (which I don't think is _actually_ needed, but I left it for Kafka due to how its getEarliest/Latest sequence number works by getting current position, seeking to earliest or latest, and then seeking back to current position)
It seems like if reworked we could probably remove the need to assign partitions from the supervisor too, and further simplify things.
----------------------------------------------------------------
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
[GitHub] [druid] clintropolis commented on a change in pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
Posted by GitBox <gi...@apache.org>.
clintropolis commented on a change in pull request #9819:
URL: https://github.com/apache/druid/pull/9819#discussion_r421293489
##########
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:
Woops, this was a rename done with intellij that caught this method in the crossfire, I only meant to rename `getPartitionResourcesTimeLag()` since it does actually get the lag from the `PartitionResource` collection. Fixed.
----------------------------------------------------------------
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
[GitHub] [druid] jihoonson commented on a change in pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
Posted by GitBox <gi...@apache.org>.
jihoonson commented on a change in pull request #9819:
URL: https://github.com/apache/druid/pull/9819#discussion_r426045765
##########
File path: extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -847,36 +881,46 @@ private void checkIfClosed()
}
/**
- * Returns an array with the content between the position and limit of "buffer". This may be the buffer's backing
- * array itself. Does not modify position or limit of the buffer.
+ * This method must be called before a seek operation ({@link #seek}, {@link #seekToLatest}, or
+ * {@link #seekToEarliest}).
+ *
+ * When called, it will nuke the {@link #scheduledExec} that is shared by all {@link PartitionResource}, filters
+ * records from the buffer for partitions which will have a seek operation performed, and stops background fetch for
+ * each {@link PartitionResource} to prepare for the seek. If background fetch is not currently running, the
+ * {@link #scheduledExec} will not be re-created.
*/
- private static byte[] toByteArray(final ByteBuffer buffer)
+ private void filterBufferAndResetBackgroundFetch(Set<StreamPartition<String>> partitions) throws InterruptedException
{
- if (buffer.hasArray()
- && buffer.arrayOffset() == 0
- && buffer.position() == 0
- && buffer.array().length == buffer.limit()) {
- return buffer.array();
- } else {
- final byte[] retVal = new byte[buffer.remaining()];
- buffer.duplicate().get(retVal);
- return retVal;
- }
- }
+ checkIfClosed();
Review comment:
Yeah, seems it should check here to avoid unnecessary seek (this method is called in seek()`.
##########
File path: extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -847,36 +881,46 @@ private void checkIfClosed()
}
/**
- * Returns an array with the content between the position and limit of "buffer". This may be the buffer's backing
- * array itself. Does not modify position or limit of the buffer.
+ * This method must be called before a seek operation ({@link #seek}, {@link #seekToLatest}, or
+ * {@link #seekToEarliest}).
+ *
+ * When called, it will nuke the {@link #scheduledExec} that is shared by all {@link PartitionResource}, filters
+ * records from the buffer for partitions which will have a seek operation performed, and stops background fetch for
+ * each {@link PartitionResource} to prepare for the seek. If background fetch is not currently running, the
+ * {@link #scheduledExec} will not be re-created.
*/
- private static byte[] toByteArray(final ByteBuffer buffer)
+ private void filterBufferAndResetBackgroundFetch(Set<StreamPartition<String>> partitions) throws InterruptedException
{
- if (buffer.hasArray()
- && buffer.arrayOffset() == 0
- && buffer.position() == 0
- && buffer.array().length == buffer.limit()) {
- return buffer.array();
- } else {
- final byte[] retVal = new byte[buffer.remaining()];
- buffer.duplicate().get(retVal);
- return retVal;
- }
- }
+ checkIfClosed();
Review comment:
Yeah, seems it should check here to avoid unnecessary seek (this method is called in `seek()`).
----------------------------------------------------------------
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
[GitHub] [druid] clintropolis commented on a change in pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
Posted by GitBox <gi...@apache.org>.
clintropolis commented on a change in pull request #9819:
URL: https://github.com/apache/druid/pull/9819#discussion_r425673901
##########
File path: extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -847,36 +881,46 @@ private void checkIfClosed()
}
/**
- * Returns an array with the content between the position and limit of "buffer". This may be the buffer's backing
- * array itself. Does not modify position or limit of the buffer.
+ * This method must be called before a seek operation ({@link #seek}, {@link #seekToLatest}, or
+ * {@link #seekToEarliest}).
+ *
+ * When called, it will nuke the {@link #scheduledExec} that is shared by all {@link PartitionResource}, filters
+ * records from the buffer for partitions which will have a seek operation performed, and stops background fetch for
+ * each {@link PartitionResource} to prepare for the seek. If background fetch is not currently running, the
+ * {@link #scheduledExec} will not be re-created.
*/
- private static byte[] toByteArray(final ByteBuffer buffer)
+ private void filterBufferAndResetBackgroundFetch(Set<StreamPartition<String>> partitions) throws InterruptedException
{
- if (buffer.hasArray()
- && buffer.arrayOffset() == 0
- && buffer.position() == 0
- && buffer.array().length == buffer.limit()) {
- return buffer.array();
- } else {
- final byte[] retVal = new byte[buffer.remaining()];
- buffer.duplicate().get(retVal);
- return retVal;
- }
- }
+ checkIfClosed();
Review comment:
Added check for `backgroundFetchEnabled` to the if statement, but `checkIfClose` will still be called, is that cool? (since you put the comment on this line)
----------------------------------------------------------------
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
[GitHub] [druid] jihoonson commented on a change in pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
Posted by GitBox <gi...@apache.org>.
jihoonson commented on a change in pull request #9819:
URL: https://github.com/apache/druid/pull/9819#discussion_r425349893
##########
File path: extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java
##########
@@ -334,16 +334,35 @@ protected boolean useExclusiveStartSequenceNumberForNonFirstSequence()
}
@Override
- protected void updateLatestSequenceFromStream(
- RecordSupplier<Integer, Long> recordSupplier,
- Set<StreamPartition<Integer>> partitions
- )
+ protected void updatePartitionLagFromStream()
Review comment:
Should this method compute the lag instead of `latestSequenceFromStream`? This will modify the current behavior since there could be some difference between computed lag and actual lag as `highestCurrentOffsets` will be computed periodically. However, I think it would be fine since 1) the behavior of this method is consistent across Kafka and Kinesis and 2) the lag metric doesn't have to be very real time.
##########
File path: extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -847,36 +881,46 @@ private void checkIfClosed()
}
/**
- * Returns an array with the content between the position and limit of "buffer". This may be the buffer's backing
- * array itself. Does not modify position or limit of the buffer.
+ * This method must be called before a seek operation ({@link #seek}, {@link #seekToLatest}, or
+ * {@link #seekToEarliest}).
+ *
+ * When called, it will nuke the {@link #scheduledExec} that is shared by all {@link PartitionResource}, filters
+ * records from the buffer for partitions which will have a seek operation performed, and stops background fetch for
+ * each {@link PartitionResource} to prepare for the seek. If background fetch is not currently running, the
+ * {@link #scheduledExec} will not be re-created.
*/
- private static byte[] toByteArray(final ByteBuffer buffer)
+ private void filterBufferAndResetBackgroundFetch(Set<StreamPartition<String>> partitions) throws InterruptedException
{
- if (buffer.hasArray()
- && buffer.arrayOffset() == 0
- && buffer.position() == 0
- && buffer.array().length == buffer.limit()) {
- return buffer.array();
- } else {
- final byte[] retVal = new byte[buffer.remaining()];
- buffer.duplicate().get(retVal);
- return retVal;
- }
- }
+ checkIfClosed();
Review comment:
Hmm, would it be nice if `backgroundFetchEnabled` is checked in this method or in the callers of this method to make it clear that this method will do nothing if `backgroundFetchEnabled` = false?
##########
File path: extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -665,180 +694,185 @@ 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 = getPartitionTimeLag(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)
+ @VisibleForTesting
+ public int bufferSize()
{
- Map<String, Long> partitionLag = Maps.newHashMapWithExpectedSize(currentOffsets.size());
- for (Map.Entry<StreamPartition<String>, PartitionResource> partition : partitionResources.entrySet()) {
- final String partitionId = partition.getKey().getPartitionId();
- partitionLag.put(partitionId, partition.getValue().getPartitionTimeLag(currentOffsets.get(partitionId)));
- }
- return partitionLag;
+ return records.size();
}
- private void seekInternal(StreamPartition<String> partition, String sequenceNumber, ShardIteratorType iteratorEnum)
+ @VisibleForTesting
+ public boolean isBackgroundFetchRunning()
{
- PartitionResource resource = partitionResources.get(partition);
- if (resource == null) {
- throw new ISE("Partition [%s] has not been assigned", partition);
- }
-
- log.debug(
- "Seeking partition [%s] to [%s]",
- partition.getPartitionId(),
- sequenceNumber != null ? sequenceNumber : iteratorEnum.toString()
- );
-
- resource.shardIterator = wrapExceptions(() -> kinesis.getShardIterator(
- partition.getStream(),
- partition.getPartitionId(),
- iteratorEnum.toString(),
- sequenceNumber
- ).getShardIterator());
-
- checkPartitionsStarted = true;
+ return partitionsFetchStarted.get();
}
- private void filterBufferAndResetFetchRunnable(Set<StreamPartition<String>> partitions) throws InterruptedException
+ /**
+ * 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)
{
- 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 e;
+ PartitionResource resource = partitionResources.get(partition);
+ if (resource == null) {
+ throw new ISE("Partition [%s] has not been assigned", partition);
}
-
- scheduledExec = Executors.newScheduledThreadPool(
- fetchThreads,
- Execs.makeThreadFactory("KinesisRecordSupplier-Worker-%d")
- );
-
- // filter records in buffer and only retain ones whose partition was not seeked
- BlockingQueue<OrderedPartitionableRecord<String, String>> newQ = new LinkedBlockingQueue<>(recordBufferSize);
-
- records.stream()
- .filter(x -> !partitions.contains(x.getStreamPartition()))
- .forEachOrdered(newQ::offer);
-
- records = newQ;
-
- // restart fetching threads
- partitionResources.values().forEach(x -> x.started = false);
- checkPartitionsStarted = true;
- }
-
- @Nullable
- private String getSequenceNumberInternal(StreamPartition<String> partition, ShardIteratorType iteratorEnum)
- {
- return wrapExceptions(() -> getSequenceNumberInternal(
- partition,
- kinesis.getShardIterator(partition.getStream(), partition.getPartitionId(), iteratorEnum.toString())
- .getShardIterator()
- ));
+ resource.seek(iteratorEnum, sequenceNumber);
}
+ /**
+ * 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 getSequenceNumberInternal(StreamPartition<String> partition, String shardIterator)
+ private String getSequenceNumber(StreamPartition<String> partition, ShardIteratorType iteratorEnum)
{
- 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(1000));
- }
- 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."
- );
+ 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 {
- Thread.sleep(PROVISIONED_THROUGHPUT_EXCEEDED_BACKOFF_MS);
- continue;
+ // 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 (InterruptedException e1) {
- log.warn(e1, "Thread interrupted!");
- Thread.currentThread().interrupt();
- break;
+ 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);
Review comment:
nit: wondering if it's better to use fuzzy exponential backoff using `RetryUtils`. I guess it could be better by avoiding retries which could be repeated at the same time.
##########
File path: extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -112,100 +145,75 @@ private static boolean isServiceExceptionRecoverable(AmazonServiceException ex)
// to indicate that this shard has no more records to read
@Nullable
private volatile String shardIterator;
- private volatile boolean started;
- private volatile boolean stopRequested;
-
private volatile long currentLagMillis;
- PartitionResource(StreamPartition<String> streamPartition)
+ private final AtomicBoolean fetchStarted = new AtomicBoolean();
+
+ private PartitionResource(StreamPartition<String> streamPartition)
{
this.streamPartition = streamPartition;
}
- void startBackgroundFetch()
+ private void startBackgroundFetch()
{
- if (started) {
+ if (!backgroundFetchEnabled) {
return;
}
+ // if seek has been called
+ if (shardIterator == null) {
+ log.warn(
+ "Skipping background fetch for stream[%s] partition[%s] since seek has not been called for this partition",
+ streamPartition.getStream(),
+ streamPartition.getPartitionId()
+ );
+ return;
+ }
+ if (fetchStarted.compareAndSet(false, true)) {
+ log.info(
+ "Starting scheduled fetch for stream[%s] partition[%s]",
+ streamPartition.getStream(),
+ streamPartition.getPartitionId()
+ );
- log.info(
- "Starting scheduled fetch runnable for stream[%s] partition[%s]",
- streamPartition.getStream(),
- streamPartition.getPartitionId()
- );
-
- stopRequested = false;
- started = true;
-
- rescheduleRunnable(fetchDelayMillis);
- }
-
- void stopBackgroundFetch()
- {
- log.info(
- "Stopping scheduled fetch runnable for stream[%s] partition[%s]",
- streamPartition.getStream(),
- streamPartition.getPartitionId()
- );
- stopRequested = true;
+ scheduleBackgroundFetch(fetchDelayMillis);
+ }
}
- long getPartitionTimeLag()
+ private void stopBackgroundFetch()
{
- return currentLagMillis;
+ if (fetchStarted.compareAndSet(true, false)) {
+ log.info(
Review comment:
Seems worth canceling a `ScheduledFuture` for `fetchRecords()` here since it can be called in `assign()`.
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
##########
@@ -1998,16 +2009,37 @@ private boolean updatePartitionDataFromStream()
partitionGroup.add(partitionId);
if (partitionOffsets.putIfAbsent(partitionId, getNotSetMarker()) == null) {
- log.info(
+ log.debug(
"New partition [%s] discovered for stream [%s], added to task group [%d]",
partitionId,
ioConfig.getStream(),
taskGroupId
);
+
+ newlyDiscovered.compute(taskGroupId, (groupId, partitions) -> {
+ if (partitions == null) {
+ partitions = new ArrayList<>();
+ }
+ partitions.add(partitionId);
Review comment:
This seems duplicate. Also `compute()` will execute the function and replace the value even when the key exists. You can do `newlyDiscovered.computeIfAbsent(taskGroupId, ArrayList::new).add(partitionId);`.
----------------------------------------------------------------
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
[GitHub] [druid] jihoonson commented on a change in pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
Posted by GitBox <gi...@apache.org>.
jihoonson commented on a change in pull request #9819:
URL: https://github.com/apache/druid/pull/9819#discussion_r426045717
##########
File path: extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java
##########
@@ -334,16 +334,35 @@ protected boolean useExclusiveStartSequenceNumberForNonFirstSequence()
}
@Override
- protected void updateLatestSequenceFromStream(
- RecordSupplier<Integer, Long> recordSupplier,
- Set<StreamPartition<Integer>> partitions
- )
+ protected void updatePartitionLagFromStream()
Review comment:
Sounds good to me.
##########
File path: extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -847,36 +881,46 @@ private void checkIfClosed()
}
/**
- * Returns an array with the content between the position and limit of "buffer". This may be the buffer's backing
- * array itself. Does not modify position or limit of the buffer.
+ * This method must be called before a seek operation ({@link #seek}, {@link #seekToLatest}, or
+ * {@link #seekToEarliest}).
+ *
+ * When called, it will nuke the {@link #scheduledExec} that is shared by all {@link PartitionResource}, filters
+ * records from the buffer for partitions which will have a seek operation performed, and stops background fetch for
+ * each {@link PartitionResource} to prepare for the seek. If background fetch is not currently running, the
+ * {@link #scheduledExec} will not be re-created.
*/
- private static byte[] toByteArray(final ByteBuffer buffer)
+ private void filterBufferAndResetBackgroundFetch(Set<StreamPartition<String>> partitions) throws InterruptedException
{
- if (buffer.hasArray()
- && buffer.arrayOffset() == 0
- && buffer.position() == 0
- && buffer.array().length == buffer.limit()) {
- return buffer.array();
- } else {
- final byte[] retVal = new byte[buffer.remaining()];
- buffer.duplicate().get(retVal);
- return retVal;
- }
- }
+ checkIfClosed();
Review comment:
Yeah, seems it should check to avoid unnecessary seek.
##########
File path: extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -665,180 +694,185 @@ 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 = getPartitionTimeLag(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)
+ @VisibleForTesting
+ public int bufferSize()
{
- Map<String, Long> partitionLag = Maps.newHashMapWithExpectedSize(currentOffsets.size());
- for (Map.Entry<StreamPartition<String>, PartitionResource> partition : partitionResources.entrySet()) {
- final String partitionId = partition.getKey().getPartitionId();
- partitionLag.put(partitionId, partition.getValue().getPartitionTimeLag(currentOffsets.get(partitionId)));
- }
- return partitionLag;
+ return records.size();
}
- private void seekInternal(StreamPartition<String> partition, String sequenceNumber, ShardIteratorType iteratorEnum)
+ @VisibleForTesting
+ public boolean isBackgroundFetchRunning()
{
- PartitionResource resource = partitionResources.get(partition);
- if (resource == null) {
- throw new ISE("Partition [%s] has not been assigned", partition);
- }
-
- log.debug(
- "Seeking partition [%s] to [%s]",
- partition.getPartitionId(),
- sequenceNumber != null ? sequenceNumber : iteratorEnum.toString()
- );
-
- resource.shardIterator = wrapExceptions(() -> kinesis.getShardIterator(
- partition.getStream(),
- partition.getPartitionId(),
- iteratorEnum.toString(),
- sequenceNumber
- ).getShardIterator());
-
- checkPartitionsStarted = true;
+ return partitionsFetchStarted.get();
}
- private void filterBufferAndResetFetchRunnable(Set<StreamPartition<String>> partitions) throws InterruptedException
+ /**
+ * 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)
{
- 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 e;
+ PartitionResource resource = partitionResources.get(partition);
+ if (resource == null) {
+ throw new ISE("Partition [%s] has not been assigned", partition);
}
-
- scheduledExec = Executors.newScheduledThreadPool(
- fetchThreads,
- Execs.makeThreadFactory("KinesisRecordSupplier-Worker-%d")
- );
-
- // filter records in buffer and only retain ones whose partition was not seeked
- BlockingQueue<OrderedPartitionableRecord<String, String>> newQ = new LinkedBlockingQueue<>(recordBufferSize);
-
- records.stream()
- .filter(x -> !partitions.contains(x.getStreamPartition()))
- .forEachOrdered(newQ::offer);
-
- records = newQ;
-
- // restart fetching threads
- partitionResources.values().forEach(x -> x.started = false);
- checkPartitionsStarted = true;
- }
-
- @Nullable
- private String getSequenceNumberInternal(StreamPartition<String> partition, ShardIteratorType iteratorEnum)
- {
- return wrapExceptions(() -> getSequenceNumberInternal(
- partition,
- kinesis.getShardIterator(partition.getStream(), partition.getPartitionId(), iteratorEnum.toString())
- .getShardIterator()
- ));
+ resource.seek(iteratorEnum, sequenceNumber);
}
+ /**
+ * 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 getSequenceNumberInternal(StreamPartition<String> partition, String shardIterator)
+ private String getSequenceNumber(StreamPartition<String> partition, ShardIteratorType iteratorEnum)
{
- 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(1000));
- }
- 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."
- );
+ 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 {
- Thread.sleep(PROVISIONED_THROUGHPUT_EXCEEDED_BACKOFF_MS);
- continue;
+ // 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 (InterruptedException e1) {
- log.warn(e1, "Thread interrupted!");
- Thread.currentThread().interrupt();
- break;
+ 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);
Review comment:
I guess we can add another mode to `RetryUtils.retry()` that checks the timeout instead of retryCount, but the current way looks ok to me.
----------------------------------------------------------------
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