| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.distributedlog; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Optional; |
| import com.google.common.base.Stopwatch; |
| import com.google.common.base.Ticker; |
| import org.apache.distributedlog.exceptions.DLIllegalStateException; |
| import org.apache.distributedlog.exceptions.DLInterruptedException; |
| import org.apache.distributedlog.exceptions.EndOfStreamException; |
| import org.apache.distributedlog.exceptions.IdleReaderException; |
| import org.apache.distributedlog.exceptions.LogNotFoundException; |
| import org.apache.distributedlog.exceptions.ReadCancelledException; |
| import org.apache.distributedlog.exceptions.UnexpectedException; |
| import org.apache.distributedlog.util.OrderedScheduler; |
| import org.apache.distributedlog.util.Utils; |
| import com.twitter.util.Future; |
| import com.twitter.util.FutureEventListener; |
| import com.twitter.util.Promise; |
| import com.twitter.util.Throw; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.concurrent.ConcurrentLinkedQueue; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.ScheduledFuture; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.concurrent.atomic.AtomicReference; |
| |
| import org.apache.bookkeeper.stats.Counter; |
| import org.apache.bookkeeper.stats.OpStatsLogger; |
| import org.apache.bookkeeper.stats.StatsLogger; |
| import org.apache.bookkeeper.versioning.Versioned; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import scala.Function1; |
| import scala.runtime.AbstractFunction1; |
| import scala.runtime.BoxedUnit; |
| |
| /** |
| * BookKeeper based {@link AsyncLogReader} implementation. |
| * |
| * <h3>Metrics</h3> |
| * All the metrics are exposed under `async_reader`. |
| * <ul> |
| * <li> `async_reader`/future_set: opstats. time spent on satisfying futures of read requests. |
| * if it is high, it means that the caller takes time on processing the result of read requests. |
| * The side effect is blocking consequent reads. |
| * <li> `async_reader`/schedule: opstats. time spent on scheduling next reads. |
| * <li> `async_reader`/background_read: opstats. time spent on background reads. |
| * <li> `async_reader`/read_next_exec: opstats. time spent on executing {@link #readNext()}. |
| * <li> `async_reader`/time_between_read_next: opstats. time spent on between two consequent {@link #readNext()}. |
| * if it is high, it means that the caller is slowing down on calling {@link #readNext()}. |
| * <li> `async_reader`/delay_until_promise_satisfied: opstats. total latency for the read requests. |
| * <li> `async_reader`/idle_reader_error: counter. the number idle reader errors. |
| * </ul> |
| */ |
| class BKAsyncLogReader implements AsyncLogReader, Runnable, AsyncNotification { |
| static final Logger LOG = LoggerFactory.getLogger(BKAsyncLogReader.class); |
| |
| private static final Function1<List<LogRecordWithDLSN>, LogRecordWithDLSN> READ_NEXT_MAP_FUNCTION = |
| new AbstractFunction1<List<LogRecordWithDLSN>, LogRecordWithDLSN>() { |
| @Override |
| public LogRecordWithDLSN apply(List<LogRecordWithDLSN> records) { |
| return records.get(0); |
| } |
| }; |
| |
| private final String streamName; |
| protected final BKDistributedLogManager bkDistributedLogManager; |
| protected final BKLogReadHandler readHandler; |
| private final AtomicReference<Throwable> lastException = new AtomicReference<Throwable>(); |
| private final OrderedScheduler scheduler; |
| private final ConcurrentLinkedQueue<PendingReadRequest> pendingRequests = new ConcurrentLinkedQueue<PendingReadRequest>(); |
| private final Object scheduleLock = new Object(); |
| private final AtomicLong scheduleCount = new AtomicLong(0); |
| final private Stopwatch scheduleDelayStopwatch; |
| final private Stopwatch readNextDelayStopwatch; |
| private DLSN startDLSN; |
| private ReadAheadEntryReader readAheadReader = null; |
| private int lastPosition = 0; |
| private final boolean positionGapDetectionEnabled; |
| private final int idleErrorThresholdMillis; |
| final ScheduledFuture<?> idleReaderTimeoutTask; |
| private ScheduledFuture<?> backgroundScheduleTask = null; |
| // last process time |
| private final Stopwatch lastProcessTime; |
| |
| protected Promise<Void> closeFuture = null; |
| |
| private boolean lockStream = false; |
| |
| private final boolean returnEndOfStreamRecord; |
| |
| private final Runnable BACKGROUND_READ_SCHEDULER = new Runnable() { |
| @Override |
| public void run() { |
| synchronized (scheduleLock) { |
| backgroundScheduleTask = null; |
| } |
| scheduleBackgroundRead(); |
| } |
| }; |
| |
| // State |
| private Entry.Reader currentEntry = null; |
| private LogRecordWithDLSN nextRecord = null; |
| |
| // Failure Injector |
| private boolean disableProcessingReadRequests = false; |
| |
| // Stats |
| private final OpStatsLogger readNextExecTime; |
| private final OpStatsLogger delayUntilPromiseSatisfied; |
| private final OpStatsLogger timeBetweenReadNexts; |
| private final OpStatsLogger futureSetLatency; |
| private final OpStatsLogger scheduleLatency; |
| private final OpStatsLogger backgroundReaderRunTime; |
| private final Counter idleReaderCheckCount; |
| private final Counter idleReaderCheckIdleReadRequestCount; |
| private final Counter idleReaderCheckIdleReadAheadCount; |
| private final Counter idleReaderError; |
| |
| private class PendingReadRequest { |
| private final Stopwatch enqueueTime; |
| private final int numEntries; |
| private final List<LogRecordWithDLSN> records; |
| private final Promise<List<LogRecordWithDLSN>> promise; |
| private final long deadlineTime; |
| private final TimeUnit deadlineTimeUnit; |
| |
| PendingReadRequest(int numEntries, |
| long deadlineTime, |
| TimeUnit deadlineTimeUnit) { |
| this.numEntries = numEntries; |
| this.enqueueTime = Stopwatch.createStarted(); |
| // optimize the space usage for single read. |
| if (numEntries == 1) { |
| this.records = new ArrayList<LogRecordWithDLSN>(1); |
| } else { |
| this.records = new ArrayList<LogRecordWithDLSN>(); |
| } |
| this.promise = new Promise<List<LogRecordWithDLSN>>(); |
| this.deadlineTime = deadlineTime; |
| this.deadlineTimeUnit = deadlineTimeUnit; |
| } |
| |
| Promise<List<LogRecordWithDLSN>> getPromise() { |
| return promise; |
| } |
| |
| long elapsedSinceEnqueue(TimeUnit timeUnit) { |
| return enqueueTime.elapsed(timeUnit); |
| } |
| |
| void setException(Throwable throwable) { |
| Stopwatch stopwatch = Stopwatch.createStarted(); |
| if (promise.updateIfEmpty(new Throw<List<LogRecordWithDLSN>>(throwable))) { |
| futureSetLatency.registerFailedEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS)); |
| delayUntilPromiseSatisfied.registerFailedEvent(enqueueTime.elapsed(TimeUnit.MICROSECONDS)); |
| } |
| } |
| |
| boolean hasReadRecords() { |
| return records.size() > 0; |
| } |
| |
| boolean hasReadEnoughRecords() { |
| return records.size() >= numEntries; |
| } |
| |
| long getRemainingWaitTime() { |
| if (deadlineTime <= 0L) { |
| return 0L; |
| } |
| return deadlineTime - elapsedSinceEnqueue(deadlineTimeUnit); |
| } |
| |
| void addRecord(LogRecordWithDLSN record) { |
| records.add(record); |
| } |
| |
| void complete() { |
| if (LOG.isTraceEnabled()) { |
| LOG.trace("{} : Satisfied promise with {} records", readHandler.getFullyQualifiedName(), records.size()); |
| } |
| delayUntilPromiseSatisfied.registerSuccessfulEvent(enqueueTime.stop().elapsed(TimeUnit.MICROSECONDS)); |
| Stopwatch stopwatch = Stopwatch.createStarted(); |
| promise.setValue(records); |
| futureSetLatency.registerSuccessfulEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS)); |
| } |
| } |
| |
| BKAsyncLogReader(BKDistributedLogManager bkdlm, |
| OrderedScheduler scheduler, |
| DLSN startDLSN, |
| Optional<String> subscriberId, |
| boolean returnEndOfStreamRecord, |
| StatsLogger statsLogger) { |
| this.streamName = bkdlm.getStreamName(); |
| this.bkDistributedLogManager = bkdlm; |
| this.scheduler = scheduler; |
| this.readHandler = bkDistributedLogManager.createReadHandler(subscriberId, |
| this, true); |
| LOG.debug("Starting async reader at {}", startDLSN); |
| this.startDLSN = startDLSN; |
| this.scheduleDelayStopwatch = Stopwatch.createUnstarted(); |
| this.readNextDelayStopwatch = Stopwatch.createStarted(); |
| this.positionGapDetectionEnabled = bkdlm.getConf().getPositionGapDetectionEnabled(); |
| this.idleErrorThresholdMillis = bkdlm.getConf().getReaderIdleErrorThresholdMillis(); |
| this.returnEndOfStreamRecord = returnEndOfStreamRecord; |
| |
| // Stats |
| StatsLogger asyncReaderStatsLogger = statsLogger.scope("async_reader"); |
| futureSetLatency = asyncReaderStatsLogger.getOpStatsLogger("future_set"); |
| scheduleLatency = asyncReaderStatsLogger.getOpStatsLogger("schedule"); |
| backgroundReaderRunTime = asyncReaderStatsLogger.getOpStatsLogger("background_read"); |
| readNextExecTime = asyncReaderStatsLogger.getOpStatsLogger("read_next_exec"); |
| timeBetweenReadNexts = asyncReaderStatsLogger.getOpStatsLogger("time_between_read_next"); |
| delayUntilPromiseSatisfied = asyncReaderStatsLogger.getOpStatsLogger("delay_until_promise_satisfied"); |
| idleReaderError = asyncReaderStatsLogger.getCounter("idle_reader_error"); |
| idleReaderCheckCount = asyncReaderStatsLogger.getCounter("idle_reader_check_total"); |
| idleReaderCheckIdleReadRequestCount = asyncReaderStatsLogger.getCounter("idle_reader_check_idle_read_requests"); |
| idleReaderCheckIdleReadAheadCount = asyncReaderStatsLogger.getCounter("idle_reader_check_idle_readahead"); |
| |
| // Lock the stream if requested. The lock will be released when the reader is closed. |
| this.lockStream = false; |
| this.idleReaderTimeoutTask = scheduleIdleReaderTaskIfNecessary(); |
| this.lastProcessTime = Stopwatch.createStarted(); |
| } |
| |
| private ScheduledFuture<?> scheduleIdleReaderTaskIfNecessary() { |
| if (idleErrorThresholdMillis < Integer.MAX_VALUE) { |
| // Dont run the task more than once every seconds (for sanity) |
| long period = Math.max(idleErrorThresholdMillis / 10, 1000); |
| // Except when idle reader threshold is less than a second (tests?) |
| period = Math.min(period, idleErrorThresholdMillis / 5); |
| |
| return scheduler.scheduleAtFixedRate(streamName, new Runnable() { |
| @Override |
| public void run() { |
| PendingReadRequest nextRequest = pendingRequests.peek(); |
| |
| idleReaderCheckCount.inc(); |
| if (null == nextRequest) { |
| return; |
| } |
| |
| idleReaderCheckIdleReadRequestCount.inc(); |
| if (nextRequest.elapsedSinceEnqueue(TimeUnit.MILLISECONDS) < idleErrorThresholdMillis) { |
| return; |
| } |
| |
| ReadAheadEntryReader readAheadReader = getReadAheadReader(); |
| |
| // read request has been idle |
| // - cache has records but read request are idle, |
| // that means notification was missed between readahead and reader. |
| // - cache is empty and readahead is idle (no records added for a long time) |
| idleReaderCheckIdleReadAheadCount.inc(); |
| try { |
| if (null == readAheadReader || (!hasMoreRecords() && |
| readAheadReader.isReaderIdle(idleErrorThresholdMillis, TimeUnit.MILLISECONDS))) { |
| markReaderAsIdle(); |
| return; |
| } else if (lastProcessTime.elapsed(TimeUnit.MILLISECONDS) > idleErrorThresholdMillis) { |
| markReaderAsIdle();; |
| } |
| } catch (IOException e) { |
| setLastException(e); |
| return; |
| } |
| } |
| }, period, period, TimeUnit.MILLISECONDS); |
| } |
| return null; |
| } |
| |
| synchronized ReadAheadEntryReader getReadAheadReader() { |
| return readAheadReader; |
| } |
| |
| void cancelIdleReaderTask() { |
| // Do this after we have checked that the reader was not previously closed |
| try { |
| if (null != idleReaderTimeoutTask) { |
| idleReaderTimeoutTask.cancel(true); |
| } |
| } catch (Exception exc) { |
| LOG.info("{}: Failed to cancel the background idle reader timeout task", readHandler.getFullyQualifiedName()); |
| } |
| } |
| |
| private void markReaderAsIdle() { |
| idleReaderError.inc(); |
| IdleReaderException ire = new IdleReaderException("Reader on stream " |
| + readHandler.getFullyQualifiedName() |
| + " is idle for " + idleErrorThresholdMillis +" ms"); |
| setLastException(ire); |
| // cancel all pending reads directly rather than notifying on error |
| // because idle reader could happen on idle read requests that usually means something wrong |
| // in scheduling reads |
| cancelAllPendingReads(ire); |
| } |
| |
| protected synchronized void setStartDLSN(DLSN fromDLSN) throws UnexpectedException { |
| if (null != readAheadReader) { |
| throw new UnexpectedException("Could't reset from dlsn after reader already starts reading."); |
| } |
| startDLSN = fromDLSN; |
| } |
| |
| @VisibleForTesting |
| public synchronized DLSN getStartDLSN() { |
| return startDLSN; |
| } |
| |
| public Future<Void> lockStream() { |
| this.lockStream = true; |
| return readHandler.lockStream(); |
| } |
| |
| private boolean checkClosedOrInError(String operation) { |
| if (null == lastException.get()) { |
| try { |
| if (null != readHandler && null != getReadAheadReader()) { |
| getReadAheadReader().checkLastException(); |
| } |
| |
| bkDistributedLogManager.checkClosedOrInError(operation); |
| } catch (IOException exc) { |
| setLastException(exc); |
| } |
| } |
| |
| if (lockStream) { |
| try { |
| readHandler.checkReadLock(); |
| } catch (IOException ex) { |
| setLastException(ex); |
| } |
| } |
| |
| if (null != lastException.get()) { |
| LOG.trace("Cancelling pending reads"); |
| cancelAllPendingReads(lastException.get()); |
| return true; |
| } |
| |
| return false; |
| } |
| |
| private void setLastException(IOException exc) { |
| lastException.compareAndSet(null, exc); |
| } |
| |
| @Override |
| public String getStreamName() { |
| return streamName; |
| } |
| |
| /** |
| * @return A promise that when satisfied will contain the Log Record with its DLSN. |
| */ |
| @Override |
| public synchronized Future<LogRecordWithDLSN> readNext() { |
| return readInternal(1, 0, TimeUnit.MILLISECONDS).map(READ_NEXT_MAP_FUNCTION); |
| } |
| |
| public synchronized Future<List<LogRecordWithDLSN>> readBulk(int numEntries) { |
| return readInternal(numEntries, 0, TimeUnit.MILLISECONDS); |
| } |
| |
| @Override |
| public synchronized Future<List<LogRecordWithDLSN>> readBulk(int numEntries, |
| long waitTime, |
| TimeUnit timeUnit) { |
| return readInternal(numEntries, waitTime, timeUnit); |
| } |
| |
| /** |
| * Read up to <i>numEntries</i> entries. The future will be satisfied when any number of entries are |
| * ready (1 to <i>numEntries</i>). |
| * |
| * @param numEntries |
| * num entries to read |
| * @return A promise that satisfied with a non-empty list of log records with their DLSN. |
| */ |
| private synchronized Future<List<LogRecordWithDLSN>> readInternal(int numEntries, |
| long deadlineTime, |
| TimeUnit deadlineTimeUnit) { |
| timeBetweenReadNexts.registerSuccessfulEvent(readNextDelayStopwatch.elapsed(TimeUnit.MICROSECONDS)); |
| readNextDelayStopwatch.reset().start(); |
| final PendingReadRequest readRequest = new PendingReadRequest(numEntries, deadlineTime, deadlineTimeUnit); |
| |
| if (null == readAheadReader) { |
| final ReadAheadEntryReader readAheadEntryReader = this.readAheadReader = new ReadAheadEntryReader( |
| getStreamName(), |
| getStartDLSN(), |
| bkDistributedLogManager.getConf(), |
| readHandler, |
| bkDistributedLogManager.getReaderEntryStore(), |
| bkDistributedLogManager.getScheduler(), |
| Ticker.systemTicker(), |
| bkDistributedLogManager.alertStatsLogger); |
| readHandler.checkLogStreamExists().addEventListener(new FutureEventListener<Void>() { |
| @Override |
| public void onSuccess(Void value) { |
| try { |
| readHandler.registerListener(readAheadEntryReader); |
| readHandler.asyncStartFetchLogSegments() |
| .map(new AbstractFunction1<Versioned<List<LogSegmentMetadata>>, BoxedUnit>() { |
| @Override |
| public BoxedUnit apply(Versioned<List<LogSegmentMetadata>> logSegments) { |
| readAheadEntryReader.addStateChangeNotification(BKAsyncLogReader.this); |
| readAheadEntryReader.start(logSegments.getValue()); |
| return BoxedUnit.UNIT; |
| } |
| }); |
| } catch (Exception exc) { |
| notifyOnError(exc); |
| } |
| } |
| |
| @Override |
| public void onFailure(Throwable cause) { |
| notifyOnError(cause); |
| } |
| }); |
| } |
| |
| if (checkClosedOrInError("readNext")) { |
| readRequest.setException(lastException.get()); |
| } else { |
| boolean queueEmpty = pendingRequests.isEmpty(); |
| pendingRequests.add(readRequest); |
| |
| if (queueEmpty) { |
| scheduleBackgroundRead(); |
| } |
| } |
| |
| readNextExecTime.registerSuccessfulEvent(readNextDelayStopwatch.elapsed(TimeUnit.MICROSECONDS)); |
| readNextDelayStopwatch.reset().start(); |
| |
| return readRequest.getPromise(); |
| } |
| |
| public synchronized void scheduleBackgroundRead() { |
| // if the reader is already closed, we don't need to schedule background read again. |
| if (null != closeFuture) { |
| return; |
| } |
| |
| long prevCount = scheduleCount.getAndIncrement(); |
| if (0 == prevCount) { |
| scheduleDelayStopwatch.reset().start(); |
| scheduler.submit(streamName, this); |
| } |
| } |
| |
| @Override |
| public Future<Void> asyncClose() { |
| // Cancel the idle reader timeout task, interrupting if necessary |
| ReadCancelledException exception; |
| Promise<Void> closePromise; |
| synchronized (this) { |
| if (null != closeFuture) { |
| return closeFuture; |
| } |
| closePromise = closeFuture = new Promise<Void>(); |
| exception = new ReadCancelledException(readHandler.getFullyQualifiedName(), "Reader was closed"); |
| setLastException(exception); |
| } |
| |
| // Do this after we have checked that the reader was not previously closed |
| cancelIdleReaderTask(); |
| |
| synchronized (scheduleLock) { |
| if (null != backgroundScheduleTask) { |
| backgroundScheduleTask.cancel(true); |
| } |
| } |
| |
| cancelAllPendingReads(exception); |
| |
| ReadAheadEntryReader readAheadReader = getReadAheadReader(); |
| if (null != readAheadReader) { |
| readHandler.unregisterListener(readAheadReader); |
| readAheadReader.removeStateChangeNotification(this); |
| } |
| Utils.closeSequence(bkDistributedLogManager.getScheduler(), true, |
| readAheadReader, |
| readHandler |
| ).proxyTo(closePromise); |
| return closePromise; |
| } |
| |
| private void cancelAllPendingReads(Throwable throwExc) { |
| for (PendingReadRequest promise : pendingRequests) { |
| promise.setException(throwExc); |
| } |
| pendingRequests.clear(); |
| } |
| |
| synchronized boolean hasMoreRecords() throws IOException { |
| if (null == readAheadReader) { |
| return false; |
| } |
| if (readAheadReader.getNumCachedEntries() > 0 || null != nextRecord) { |
| return true; |
| } else if (null != currentEntry) { |
| nextRecord = currentEntry.nextRecord(); |
| return null != nextRecord; |
| } |
| return false; |
| } |
| |
| private synchronized LogRecordWithDLSN readNextRecord() throws IOException { |
| if (null == readAheadReader) { |
| return null; |
| } |
| if (null == currentEntry) { |
| currentEntry = readAheadReader.getNextReadAheadEntry(0L, TimeUnit.MILLISECONDS); |
| // no entry after reading from read ahead then return null |
| if (null == currentEntry) { |
| return null; |
| } |
| } |
| |
| LogRecordWithDLSN recordToReturn; |
| if (null == nextRecord) { |
| nextRecord = currentEntry.nextRecord(); |
| // no more records in current entry |
| if (null == nextRecord) { |
| currentEntry = null; |
| return readNextRecord(); |
| } |
| } |
| |
| // found a record to return and prefetch the next one |
| recordToReturn = nextRecord; |
| nextRecord = currentEntry.nextRecord(); |
| return recordToReturn; |
| } |
| |
| @Override |
| public void run() { |
| synchronized(scheduleLock) { |
| if (scheduleDelayStopwatch.isRunning()) { |
| scheduleLatency.registerSuccessfulEvent(scheduleDelayStopwatch.stop().elapsed(TimeUnit.MICROSECONDS)); |
| } |
| |
| Stopwatch runTime = Stopwatch.createStarted(); |
| int iterations = 0; |
| long scheduleCountLocal = scheduleCount.get(); |
| LOG.debug("{}: Scheduled Background Reader", readHandler.getFullyQualifiedName()); |
| while(true) { |
| if (LOG.isTraceEnabled()) { |
| LOG.trace("{}: Executing Iteration: {}", readHandler.getFullyQualifiedName(), iterations++); |
| } |
| |
| PendingReadRequest nextRequest = null; |
| synchronized(this) { |
| nextRequest = pendingRequests.peek(); |
| |
| // Queue is empty, nothing to read, return |
| if (null == nextRequest) { |
| LOG.trace("{}: Queue Empty waiting for Input", readHandler.getFullyQualifiedName()); |
| scheduleCount.set(0); |
| backgroundReaderRunTime.registerSuccessfulEvent(runTime.stop().elapsed(TimeUnit.MICROSECONDS)); |
| return; |
| } |
| |
| if (disableProcessingReadRequests) { |
| LOG.info("Reader of {} is forced to stop processing read requests", readHandler.getFullyQualifiedName()); |
| return; |
| } |
| } |
| lastProcessTime.reset().start(); |
| |
| // If the oldest pending promise is interrupted then we must mark |
| // the reader in error and abort all pending reads since we dont |
| // know the last consumed read |
| if (null == lastException.get()) { |
| if (nextRequest.getPromise().isInterrupted().isDefined()) { |
| setLastException(new DLInterruptedException("Interrupted on reading " + readHandler.getFullyQualifiedName() + " : ", |
| nextRequest.getPromise().isInterrupted().get())); |
| } |
| } |
| |
| if (checkClosedOrInError("readNext")) { |
| if (!(lastException.get().getCause() instanceof LogNotFoundException)) { |
| LOG.warn("{}: Exception", readHandler.getFullyQualifiedName(), lastException.get()); |
| } |
| backgroundReaderRunTime.registerFailedEvent(runTime.stop().elapsed(TimeUnit.MICROSECONDS)); |
| return; |
| } |
| |
| try { |
| // Fail 10% of the requests when asked to simulate errors |
| if (bkDistributedLogManager.getFailureInjector().shouldInjectErrors()) { |
| throw new IOException("Reader Simulated Exception"); |
| } |
| LogRecordWithDLSN record; |
| while (!nextRequest.hasReadEnoughRecords()) { |
| // read single record |
| do { |
| record = readNextRecord(); |
| } while (null != record && (record.isControl() || (record.getDlsn().compareTo(getStartDLSN()) < 0))); |
| if (null == record) { |
| break; |
| } else { |
| if (record.isEndOfStream() && !returnEndOfStreamRecord) { |
| setLastException(new EndOfStreamException("End of Stream Reached for " |
| + readHandler.getFullyQualifiedName())); |
| break; |
| } |
| |
| // gap detection |
| if (recordPositionsContainsGap(record, lastPosition)) { |
| bkDistributedLogManager.raiseAlert("Gap detected between records at record = {}", record); |
| if (positionGapDetectionEnabled) { |
| throw new DLIllegalStateException("Gap detected between records at record = " + record); |
| } |
| } |
| lastPosition = record.getLastPositionWithinLogSegment(); |
| |
| nextRequest.addRecord(record); |
| } |
| }; |
| } catch (IOException exc) { |
| setLastException(exc); |
| if (!(exc instanceof LogNotFoundException)) { |
| LOG.warn("{} : read with skip Exception", readHandler.getFullyQualifiedName(), lastException.get()); |
| } |
| continue; |
| } |
| |
| if (nextRequest.hasReadRecords()) { |
| long remainingWaitTime = nextRequest.getRemainingWaitTime(); |
| if (remainingWaitTime > 0 && !nextRequest.hasReadEnoughRecords()) { |
| backgroundReaderRunTime.registerSuccessfulEvent(runTime.stop().elapsed(TimeUnit.MICROSECONDS)); |
| scheduleDelayStopwatch.reset().start(); |
| scheduleCount.set(0); |
| // the request could still wait for more records |
| backgroundScheduleTask = scheduler.schedule( |
| streamName, |
| BACKGROUND_READ_SCHEDULER, |
| remainingWaitTime, |
| nextRequest.deadlineTimeUnit); |
| return; |
| } |
| |
| PendingReadRequest request = pendingRequests.poll(); |
| if (null != request && nextRequest == request) { |
| request.complete(); |
| if (null != backgroundScheduleTask) { |
| backgroundScheduleTask.cancel(true); |
| backgroundScheduleTask = null; |
| } |
| } else { |
| DLIllegalStateException ise = new DLIllegalStateException("Unexpected condition at dlsn = " |
| + nextRequest.records.get(0).getDlsn()); |
| nextRequest.setException(ise); |
| if (null != request) { |
| request.setException(ise); |
| } |
| // We should never get here as we should have exited the loop if |
| // pendingRequests were empty |
| bkDistributedLogManager.raiseAlert("Unexpected condition at dlsn = {}", |
| nextRequest.records.get(0).getDlsn()); |
| setLastException(ise); |
| } |
| } else { |
| if (0 == scheduleCountLocal) { |
| LOG.trace("Schedule count dropping to zero", lastException.get()); |
| backgroundReaderRunTime.registerSuccessfulEvent(runTime.stop().elapsed(TimeUnit.MICROSECONDS)); |
| return; |
| } |
| scheduleCountLocal = scheduleCount.decrementAndGet(); |
| } |
| } |
| } |
| } |
| |
| private boolean recordPositionsContainsGap(LogRecordWithDLSN record, long lastPosition) { |
| final boolean firstLogRecord = (1 == record.getPositionWithinLogSegment()); |
| final boolean endOfStreamRecord = record.isEndOfStream(); |
| final boolean emptyLogSegment = (0 == lastPosition); |
| final boolean positionIncreasedByOne = (record.getPositionWithinLogSegment() == (lastPosition + 1)); |
| |
| return !firstLogRecord && !endOfStreamRecord && !emptyLogSegment && |
| !positionIncreasedByOne; |
| } |
| |
| /** |
| * Triggered when the background activity encounters an exception |
| */ |
| @Override |
| public void notifyOnError(Throwable cause) { |
| if (cause instanceof IOException) { |
| setLastException((IOException) cause); |
| } else { |
| setLastException(new IOException(cause)); |
| } |
| scheduleBackgroundRead(); |
| } |
| |
| /** |
| * Triggered when the background activity completes an operation |
| */ |
| @Override |
| public void notifyOnOperationComplete() { |
| scheduleBackgroundRead(); |
| } |
| |
| @VisibleForTesting |
| void simulateErrors() { |
| bkDistributedLogManager.getFailureInjector().injectErrors(true); |
| } |
| |
| @VisibleForTesting |
| synchronized void disableReadAheadLogSegmentsNotification() { |
| readHandler.disableReadAheadLogSegmentsNotification(); |
| } |
| |
| @VisibleForTesting |
| synchronized void disableProcessingReadRequests() { |
| disableProcessingReadRequests = true; |
| } |
| } |
| |
