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/14 22:19:36 UTC
[GitHub] [druid] jihoonson commented on a change in pull request #9819: refactor SeekableStreamSupervisor usage of RecordSupplier
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