From 013a5cbdc85aa510d9d39e13549b652c74bf6833 Mon Sep 17 00:00:00 2001 From: John Roesler Date: Fri, 12 Feb 2021 00:07:03 -0600 Subject: [PATCH 1/2] Revert "KAFKA-10867: Improved task idling (#9840)" This reverts commit 4d28391480fd8c547a63af119bba67fceb5d2ede. --- .../apache/kafka/streams/StreamsConfig.java | 1 - .../internals/ActiveTaskCreator.java | 3 +- .../processor/internals/PartitionGroup.java | 148 +--------- .../processor/internals/StandbyTask.java | 6 - .../processor/internals/StreamTask.java | 68 ++--- .../processor/internals/StreamThread.java | 12 +- .../streams/processor/internals/Task.java | 6 - .../processor/internals/TaskManager.java | 3 +- .../RegexSourceIntegrationTest.java | 1 - .../internals/PartitionGroupTest.java | 258 +----------------- .../processor/internals/StreamTaskTest.java | 152 ++++++++--- .../processor/internals/TaskManagerTest.java | 6 - .../testutil/LogCaptureAppender.java | 4 - .../StreamThreadStateStoreProviderTest.java | 2 +- .../kafka/streams/TopologyTestDriver.java | 24 +- 15 files changed, 177 insertions(+), 517 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java index f7f26aadfd52a..b40a257d19043 100644 --- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java +++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java @@ -144,7 +144,6 @@ public class StreamsConfig extends AbstractConfig { private static final long EOS_DEFAULT_COMMIT_INTERVAL_MS = 100L; public static final int DUMMY_THREAD_INDEX = 1; - public static final long MAX_TASK_IDLE_MS_DISABLED = -1; /** * Prefix used to provide default topic configs to be applied when creating internal topics. diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java index 86b76e3d84ed7..482a2c5d9d175 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java @@ -242,8 +242,7 @@ private StreamTask createActiveTask(final TaskId taskId, time, stateManager, recordCollector, - context, - logContext + context ); log.trace("Created task {} with assigned partitions {}", taskId, inputPartitions); diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java index 46b3429f226e6..559e3b1838486 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java @@ -17,21 +17,16 @@ package org.apache.kafka.streams.processor.internals; import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.metrics.Sensor; -import org.apache.kafka.common.utils.LogContext; -import org.apache.kafka.streams.StreamsConfig; -import org.slf4j.Logger; import java.util.Collections; import java.util.Comparator; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Iterator; import java.util.Map; import java.util.PriorityQueue; import java.util.Set; +import java.util.Iterator; +import java.util.HashSet; import java.util.function.Function; /** @@ -58,18 +53,14 @@ */ public class PartitionGroup { - private final Logger logger; private final Map partitionQueues; - private final Sensor enforcedProcessingSensor; - private final long maxTaskIdleMs; private final Sensor recordLatenessSensor; private final PriorityQueue nonEmptyQueuesByTime; private long streamTime; private int totalBuffered; private boolean allBuffered; - private final Map fetchedLags = new HashMap<>(); - private final Map idlePartitionDeadlines = new HashMap<>(); + static class RecordInfo { RecordQueue queue; @@ -87,144 +78,15 @@ RecordQueue queue() { } } - PartitionGroup(final LogContext logContext, - final Map partitionQueues, - final Sensor recordLatenessSensor, - final Sensor enforcedProcessingSensor, - final long maxTaskIdleMs) { - this.logger = logContext.logger(PartitionGroup.class); + PartitionGroup(final Map partitionQueues, final Sensor recordLatenessSensor) { nonEmptyQueuesByTime = new PriorityQueue<>(partitionQueues.size(), Comparator.comparingLong(RecordQueue::headRecordTimestamp)); this.partitionQueues = partitionQueues; - this.enforcedProcessingSensor = enforcedProcessingSensor; - this.maxTaskIdleMs = maxTaskIdleMs; this.recordLatenessSensor = recordLatenessSensor; totalBuffered = 0; allBuffered = false; streamTime = RecordQueue.UNKNOWN; } - public void addFetchedMetadata(final TopicPartition partition, final ConsumerRecords.Metadata metadata) { - final Long lag = metadata.lag(); - if (lag != null) { - logger.trace("added fetched lag {}: {}", partition, lag); - fetchedLags.put(partition, lag); - } - } - - public boolean readyToProcess(final long wallClockTime) { - if (logger.isTraceEnabled()) { - for (final Map.Entry entry : partitionQueues.entrySet()) { - logger.trace( - "buffered/lag {}: {}/{}", - entry.getKey(), - entry.getValue().size(), - fetchedLags.get(entry.getKey()) - ); - } - } - // Log-level strategy: - // TRACE for messages that don't wait for fetches - // TRACE when we waited for a fetch and decided to wait some more, as configured - // TRACE when we are ready for processing and didn't have to enforce processing - // INFO when we enforce processing, since this has to wait for fetches AND may result in disorder - - if (maxTaskIdleMs == StreamsConfig.MAX_TASK_IDLE_MS_DISABLED) { - if (logger.isTraceEnabled() && !allBuffered && totalBuffered > 0) { - final Set bufferedPartitions = new HashSet<>(); - final Set emptyPartitions = new HashSet<>(); - for (final Map.Entry entry : partitionQueues.entrySet()) { - if (entry.getValue().isEmpty()) { - emptyPartitions.add(entry.getKey()); - } else { - bufferedPartitions.add(entry.getKey()); - } - } - logger.trace("Ready for processing because max.task.idle.ms is disabled." + - "\n\tThere may be out-of-order processing for this task as a result." + - "\n\tBuffered partitions: {}" + - "\n\tNon-buffered partitions: {}", - bufferedPartitions, - emptyPartitions); - } - return true; - } - - final Set queued = new HashSet<>(); - Map enforced = null; - - for (final Map.Entry entry : partitionQueues.entrySet()) { - final TopicPartition partition = entry.getKey(); - final RecordQueue queue = entry.getValue(); - - final Long nullableFetchedLag = fetchedLags.get(partition); - - if (!queue.isEmpty()) { - // this partition is ready for processing - idlePartitionDeadlines.remove(partition); - queued.add(partition); - } else if (nullableFetchedLag == null) { - // must wait to fetch metadata for the partition - idlePartitionDeadlines.remove(partition); - logger.trace("Waiting to fetch data for {}", partition); - return false; - } else if (nullableFetchedLag > 0L) { - // must wait to poll the data we know to be on the broker - idlePartitionDeadlines.remove(partition); - logger.trace( - "Lag for {} is currently {}, but no data is buffered locally. Waiting to buffer some records.", - partition, - nullableFetchedLag - ); - return false; - } else { - // p is known to have zero lag. wait for maxTaskIdleMs to see if more data shows up. - // One alternative would be to set the deadline to nullableMetadata.receivedTimestamp + maxTaskIdleMs - // instead. That way, we would start the idling timer as of the freshness of our knowledge about the zero - // lag instead of when we happen to run this method, but realistically it's probably a small difference - // and using wall clock time seems more intuitive for users, - // since the log message will be as of wallClockTime. - idlePartitionDeadlines.putIfAbsent(partition, wallClockTime + maxTaskIdleMs); - final long deadline = idlePartitionDeadlines.get(partition); - if (wallClockTime < deadline) { - logger.trace( - "Lag for {} is currently 0 and current time is {}. Waiting for new data to be produced for configured idle time {} (deadline is {}).", - partition, - wallClockTime, - maxTaskIdleMs, - deadline - ); - return false; - } else { - // this partition is ready for processing due to the task idling deadline passing - if (enforced == null) { - enforced = new HashMap<>(); - } - enforced.put(partition, deadline); - } - } - } - if (enforced == null) { - logger.trace("All partitions were buffered locally, so this task is ready for processing."); - return true; - } else if (queued.isEmpty()) { - logger.trace("No partitions were buffered locally, so this task is not ready for processing."); - return false; - } else { - enforcedProcessingSensor.record(1.0d, wallClockTime); - logger.info("Continuing to process although some partition timestamps were not buffered locally." + - "\n\tThere may be out-of-order processing for this task as a result." + - "\n\tPartitions with local data: {}." + - "\n\tPartitions we gave up waiting for, with their corresponding deadlines: {}." + - "\n\tConfigured max.task.idle.ms: {}." + - "\n\tCurrent wall-clock time: {}.", - queued, - enforced, - maxTaskIdleMs, - wallClockTime); - return true; - } - } - // visible for testing long partitionTimestamp(final TopicPartition partition) { final RecordQueue queue = partitionQueues.get(partition); @@ -377,7 +239,7 @@ int numBuffered() { return totalBuffered; } - boolean allPartitionsBufferedLocally() { + boolean allPartitionsBuffered() { return allBuffered; } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java index d866954668618..4efb10eb566cf 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java @@ -17,7 +17,6 @@ package org.apache.kafka.streams.processor.internals; import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.metrics.Sensor; @@ -287,11 +286,6 @@ public void addRecords(final TopicPartition partition, final Iterable e2eLatencySensors = new HashMap<>(); private final InternalProcessorContext processorContext; private final RecordQueueCreator recordQueueCreator; private StampedRecord record; + private long idleStartTimeMs; private boolean commitNeeded = false; private boolean commitRequested = false; private boolean hasPendingTxCommit = false; @@ -116,8 +118,7 @@ public StreamTask(final TaskId id, final Time time, final ProcessorStateManager stateMgr, final RecordCollector recordCollector, - final InternalProcessorContext processorContext, - final LogContext logContext) { + final InternalProcessorContext processorContext) { super( id, topology, @@ -141,6 +142,12 @@ public StreamTask(final TaskId id, this.streamsMetrics = streamsMetrics; closeTaskSensor = ThreadMetrics.closeTaskSensor(threadId, streamsMetrics); final String taskId = id.toString(); + if (streamsMetrics.version() == Version.FROM_0100_TO_24) { + final Sensor parent = ThreadMetrics.commitOverTasksSensor(threadId, streamsMetrics); + enforcedProcessingSensor = TaskMetrics.enforcedProcessingSensor(threadId, taskId, streamsMetrics, parent); + } else { + enforcedProcessingSensor = TaskMetrics.enforcedProcessingSensor(threadId, taskId, streamsMetrics); + } processRatioSensor = TaskMetrics.activeProcessRatioSensor(threadId, taskId, streamsMetrics); processLatencySensor = TaskMetrics.processLatencySensor(threadId, taskId, streamsMetrics); punctuateLatencySensor = TaskMetrics.punctuateSensor(threadId, taskId, streamsMetrics); @@ -163,30 +170,17 @@ public StreamTask(final TaskId id, streamTimePunctuationQueue = new PunctuationQueue(); systemTimePunctuationQueue = new PunctuationQueue(); + maxTaskIdleMs = config.getLong(StreamsConfig.MAX_TASK_IDLE_MS_CONFIG); maxBufferedSize = config.getInt(StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG); // initialize the consumed and committed offset cache consumedOffsets = new HashMap<>(); - recordQueueCreator = new RecordQueueCreator(this.logContext, config.defaultTimestampExtractor(), config.defaultDeserializationExceptionHandler()); + recordQueueCreator = new RecordQueueCreator(logContext, config.defaultTimestampExtractor(), config.defaultDeserializationExceptionHandler()); recordInfo = new PartitionGroup.RecordInfo(); - - final Sensor enforcedProcessingSensor; - if (streamsMetrics.version() == Version.FROM_0100_TO_24) { - final Sensor parent = ThreadMetrics.commitOverTasksSensor(threadId, streamsMetrics); - enforcedProcessingSensor = TaskMetrics.enforcedProcessingSensor(threadId, taskId, streamsMetrics, parent); - } else { - enforcedProcessingSensor = TaskMetrics.enforcedProcessingSensor(threadId, taskId, streamsMetrics); - } - final long maxTaskIdleMs = config.getLong(StreamsConfig.MAX_TASK_IDLE_MS_CONFIG); - partitionGroup = new PartitionGroup( - logContext, - createPartitionQueues(), - TaskMetrics.recordLatenessSensor(threadId, taskId, streamsMetrics), - enforcedProcessingSensor, - maxTaskIdleMs - ); + partitionGroup = new PartitionGroup(createPartitionQueues(), + TaskMetrics.recordLatenessSensor(threadId, taskId, streamsMetrics)); stateMgr.registerGlobalStateStores(topology.globalStateStores()); } @@ -242,6 +236,7 @@ public void completeRestoration() { initializeMetadata(); initializeTopology(); processorContext.initialize(); + idleStartTimeMs = RecordQueue.UNKNOWN; transitionTo(State.RUNNING); @@ -632,12 +627,7 @@ record = null; /** * An active task is processable if its buffer contains data for all of its input - * source topic partitions, or if it is enforced to be processable. - * - * Note that this method is _NOT_ idempotent, because the internal bookkeeping - * consumes the partition metadata. For example, unit tests may have to invoke - * {@link #addFetchedMetadata(TopicPartition, ConsumerRecords.Metadata)} again - * invoking this method. + * source topic partitions, or if it is enforced to be processable */ public boolean isProcessable(final long wallClockTime) { if (state() == State.CLOSED) { @@ -655,7 +645,26 @@ public boolean isProcessable(final long wallClockTime) { return false; } - return partitionGroup.readyToProcess(wallClockTime); + if (partitionGroup.allPartitionsBuffered()) { + idleStartTimeMs = RecordQueue.UNKNOWN; + return true; + } else if (partitionGroup.numBuffered() > 0) { + if (idleStartTimeMs == RecordQueue.UNKNOWN) { + idleStartTimeMs = wallClockTime; + } + + if (wallClockTime - idleStartTimeMs >= maxTaskIdleMs) { + enforcedProcessingSensor.record(1.0d, wallClockTime); + return true; + } else { + return false; + } + } else { + // there's no data in any of the topics; we should reset the enforced + // processing timer + idleStartTimeMs = RecordQueue.UNKNOWN; + return false; + } } /** @@ -922,11 +931,6 @@ public void addRecords(final TopicPartition partition, final Iterable 0) { + log.debug("Main Consumer poll completed in {} ms and fetched {} records", pollLatency, numRecords); + } pollSensor.record(pollLatency, now); - if (!records.isEmpty() || !records.metadata().isEmpty()) { + if (!records.isEmpty()) { pollRecordsSensor.record(numRecords, now); taskManager.addRecordsToTasks(records); } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java index 8e3aa0b41e854..206a5efd6a7de 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java @@ -17,7 +17,6 @@ package org.apache.kafka.streams.processor.internals; import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.TimeoutException; @@ -158,11 +157,6 @@ enum TaskType { void addRecords(TopicPartition partition, Iterable> records); - /** - * Add to this task any metadata returned from the poll. - */ - void addFetchedMetadata(TopicPartition partition, ConsumerRecords.Metadata metadata); - default boolean process(final long wallClockTime) { return false; } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java index 19e91e2be5372..3ca6876a36e85 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java @@ -970,7 +970,7 @@ int commitAll() { * @param records Records, can be null */ void addRecordsToTasks(final ConsumerRecords records) { - for (final TopicPartition partition : union(HashSet::new, records.partitions(), records.metadata().keySet())) { + for (final TopicPartition partition : records.partitions()) { final Task activeTask = tasks.activeTasksForInputPartition(partition); if (activeTask == null) { @@ -980,7 +980,6 @@ void addRecordsToTasks(final ConsumerRecords records) { } activeTask.addRecords(partition, records.records(partition)); - activeTask.addFetchedMetadata(partition, records.metadata().get(partition)); } } diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java index 58b9c19cd4aec..e3999982cfcd5 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java @@ -128,7 +128,6 @@ public void setUp() throws InterruptedException { properties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100); properties.put(ConsumerConfig.METADATA_MAX_AGE_CONFIG, "1000"); properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - properties.put(StreamsConfig.MAX_TASK_IDLE_MS_CONFIG, 0L); streamsConfiguration = StreamsTestUtils.getStreamsConfig( IntegrationTestUtils.safeUniqueTestName(RegexSourceIntegrationTest.class, new TestName()), diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/PartitionGroupTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/PartitionGroupTest.java index 09558d23854e8..d8793265f5f37 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/PartitionGroupTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/PartitionGroupTest.java @@ -17,7 +17,6 @@ package org.apache.kafka.streams.processor.internals; import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.common.MetricName; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.metrics.Metrics; @@ -30,39 +29,32 @@ import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.MockTime; import org.apache.kafka.common.utils.Time; -import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler; import org.apache.kafka.streams.processor.TimestampExtractor; -import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender; import org.apache.kafka.test.InternalMockProcessorContext; import org.apache.kafka.test.MockSourceNode; import org.apache.kafka.test.MockTimestampExtractor; -import org.hamcrest.Matchers; import org.junit.Test; import java.util.Arrays; import java.util.List; -import java.util.UUID; import static org.apache.kafka.common.utils.Utils.mkEntry; import static org.apache.kafka.common.utils.Utils.mkMap; -import static org.apache.kafka.common.utils.Utils.mkSet; import static org.hamcrest.CoreMatchers.is; +import static org.apache.kafka.common.utils.Utils.mkSet; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.equalTo; -import static org.hamcrest.Matchers.hasItem; -import static org.hamcrest.Matchers.notNullValue; import static org.hamcrest.Matchers.nullValue; +import static org.hamcrest.Matchers.notNullValue; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; public class PartitionGroupTest { - - private final long maxTaskIdleMs = StreamsConfig.MAX_TASK_IDLE_MS_DISABLED; private final LogContext logContext = new LogContext("[test] "); private final Time time = new MockTime(); private final Serializer intSerializer = new IntegerSerializer(); @@ -80,7 +72,6 @@ public class PartitionGroupTest { private final byte[] recordKey = intSerializer.serialize(null, 1); private final Metrics metrics = new Metrics(); - private final Sensor enforcedProcessingSensor = metrics.sensor(UUID.randomUUID().toString()); private final MetricName lastLatenessValue = new MetricName("record-lateness-last-value", "", "", mkMap()); @@ -459,7 +450,7 @@ public void shouldUpdatePartitionQueuesShrink() { new ConsumerRecord<>("topic", 2, 6L, recordKey, recordValue)); group.addRawRecords(partition2, list2); assertEquals(list1.size() + list2.size(), group.numBuffered()); - assertTrue(group.allPartitionsBufferedLocally()); + assertTrue(group.allPartitionsBuffered()); group.nextRecord(new PartitionGroup.RecordInfo(), time.milliseconds()); // shrink list of queues @@ -468,7 +459,7 @@ public void shouldUpdatePartitionQueuesShrink() { return null; }); - assertTrue(group.allPartitionsBufferedLocally()); // because didn't add any new partitions + assertTrue(group.allPartitionsBuffered()); // because didn't add any new partitions assertEquals(list2.size(), group.numBuffered()); assertEquals(1, group.streamTime()); assertThrows(IllegalStateException.class, () -> group.partitionTimestamp(partition1)); @@ -479,11 +470,8 @@ public void shouldUpdatePartitionQueuesShrink() { @Test public void shouldUpdatePartitionQueuesExpand() { final PartitionGroup group = new PartitionGroup( - logContext, mkMap(mkEntry(partition1, queue1)), - getValueSensor(metrics, lastLatenessValue), - enforcedProcessingSensor, - maxTaskIdleMs + getValueSensor(metrics, lastLatenessValue) ); final List> list1 = Arrays.asList( new ConsumerRecord<>("topic", 1, 1L, recordKey, recordValue), @@ -491,7 +479,7 @@ public void shouldUpdatePartitionQueuesExpand() { group.addRawRecords(partition1, list1); assertEquals(list1.size(), group.numBuffered()); - assertTrue(group.allPartitionsBufferedLocally()); + assertTrue(group.allPartitionsBuffered()); group.nextRecord(new PartitionGroup.RecordInfo(), time.milliseconds()); // expand list of queues @@ -500,7 +488,7 @@ public void shouldUpdatePartitionQueuesExpand() { return createQueue2(); }); - assertFalse(group.allPartitionsBufferedLocally()); // because added new partition + assertFalse(group.allPartitionsBuffered()); // because added new partition assertEquals(1, group.numBuffered()); assertEquals(1, group.streamTime()); assertThat(group.partitionTimestamp(partition1), equalTo(1L)); @@ -511,18 +499,15 @@ public void shouldUpdatePartitionQueuesExpand() { @Test public void shouldUpdatePartitionQueuesShrinkAndExpand() { final PartitionGroup group = new PartitionGroup( - logContext, mkMap(mkEntry(partition1, queue1)), - getValueSensor(metrics, lastLatenessValue), - enforcedProcessingSensor, - maxTaskIdleMs + getValueSensor(metrics, lastLatenessValue) ); final List> list1 = Arrays.asList( new ConsumerRecord<>("topic", 1, 1L, recordKey, recordValue), new ConsumerRecord<>("topic", 1, 5L, recordKey, recordValue)); group.addRawRecords(partition1, list1); assertEquals(list1.size(), group.numBuffered()); - assertTrue(group.allPartitionsBufferedLocally()); + assertTrue(group.allPartitionsBuffered()); group.nextRecord(new PartitionGroup.RecordInfo(), time.milliseconds()); // expand and shrink list of queues @@ -531,7 +516,7 @@ public void shouldUpdatePartitionQueuesShrinkAndExpand() { return createQueue2(); }); - assertFalse(group.allPartitionsBufferedLocally()); // because added new partition + assertFalse(group.allPartitionsBuffered()); // because added new partition assertEquals(0, group.numBuffered()); assertEquals(1, group.streamTime()); assertThrows(IllegalStateException.class, () -> group.partitionTimestamp(partition1)); @@ -539,232 +524,13 @@ public void shouldUpdatePartitionQueuesShrinkAndExpand() { assertThat(group.nextRecord(new PartitionGroup.RecordInfo(), time.milliseconds()), nullValue()); // all available records removed } - @Test - public void shouldNeverWaitIfIdlingIsDisabled() { - final PartitionGroup group = new PartitionGroup( - logContext, - mkMap( - mkEntry(partition1, queue1), - mkEntry(partition2, queue2) - ), - getValueSensor(metrics, lastLatenessValue), - enforcedProcessingSensor, - StreamsConfig.MAX_TASK_IDLE_MS_DISABLED - ); - - final List> list1 = Arrays.asList( - new ConsumerRecord<>("topic", 1, 1L, recordKey, recordValue), - new ConsumerRecord<>("topic", 1, 5L, recordKey, recordValue)); - group.addRawRecords(partition1, list1); - - assertThat(group.allPartitionsBufferedLocally(), is(false)); - try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) { - LogCaptureAppender.setClassLoggerToTrace(PartitionGroup.class); - assertThat(group.readyToProcess(0L), is(true)); - assertThat( - appender.getEvents(), - hasItem(Matchers.allOf( - Matchers.hasProperty("level", equalTo("TRACE")), - Matchers.hasProperty("message", equalTo( - "[test] Ready for processing because max.task.idle.ms is disabled.\n" + - "\tThere may be out-of-order processing for this task as a result.\n" + - "\tBuffered partitions: [topic-1]\n" + - "\tNon-buffered partitions: [topic-2]" - )) - )) - ); - } - } - - @Test - public void shouldBeReadyIfAllPartitionsAreBuffered() { - final PartitionGroup group = new PartitionGroup( - logContext, - mkMap( - mkEntry(partition1, queue1), - mkEntry(partition2, queue2) - ), - getValueSensor(metrics, lastLatenessValue), - enforcedProcessingSensor, - 0L - ); - - final List> list1 = Arrays.asList( - new ConsumerRecord<>("topic", 1, 1L, recordKey, recordValue), - new ConsumerRecord<>("topic", 1, 5L, recordKey, recordValue)); - group.addRawRecords(partition1, list1); - - final List> list2 = Arrays.asList( - new ConsumerRecord<>("topic", 2, 1L, recordKey, recordValue), - new ConsumerRecord<>("topic", 2, 5L, recordKey, recordValue)); - group.addRawRecords(partition2, list2); - - assertThat(group.allPartitionsBufferedLocally(), is(true)); - try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) { - LogCaptureAppender.setClassLoggerToTrace(PartitionGroup.class); - assertThat(group.readyToProcess(0L), is(true)); - assertThat( - appender.getEvents(), - hasItem(Matchers.allOf( - Matchers.hasProperty("level", equalTo("TRACE")), - Matchers.hasProperty("message", equalTo("[test] All partitions were buffered locally, so this task is ready for processing.")) - )) - ); - } - } - - @Test - public void shouldWaitForFetchesWhenMetadataIsIncomplete() { - final PartitionGroup group = new PartitionGroup( - logContext, - mkMap( - mkEntry(partition1, queue1), - mkEntry(partition2, queue2) - ), - getValueSensor(metrics, lastLatenessValue), - enforcedProcessingSensor, - 0L - ); - - final List> list1 = Arrays.asList( - new ConsumerRecord<>("topic", 1, 1L, recordKey, recordValue), - new ConsumerRecord<>("topic", 1, 5L, recordKey, recordValue)); - group.addRawRecords(partition1, list1); - - assertThat(group.allPartitionsBufferedLocally(), is(false)); - try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) { - LogCaptureAppender.setClassLoggerToTrace(PartitionGroup.class); - assertThat(group.readyToProcess(0L), is(false)); - assertThat( - appender.getEvents(), - hasItem(Matchers.allOf( - Matchers.hasProperty("level", equalTo("TRACE")), - Matchers.hasProperty("message", equalTo("[test] Waiting to fetch data for topic-2")) - )) - ); - } - group.addFetchedMetadata(partition2, new ConsumerRecords.Metadata(0L, 0L, 0L)); - assertThat(group.readyToProcess(0L), is(true)); - } - - @Test - public void shouldWaitForPollWhenLagIsNonzero() { - final PartitionGroup group = new PartitionGroup( - logContext, - mkMap( - mkEntry(partition1, queue1), - mkEntry(partition2, queue2) - ), - getValueSensor(metrics, lastLatenessValue), - enforcedProcessingSensor, - 0L - ); - - final List> list1 = Arrays.asList( - new ConsumerRecord<>("topic", 1, 1L, recordKey, recordValue), - new ConsumerRecord<>("topic", 1, 5L, recordKey, recordValue)); - group.addRawRecords(partition1, list1); - group.addFetchedMetadata(partition2, new ConsumerRecords.Metadata(0L, 0L, 1L)); - - assertThat(group.allPartitionsBufferedLocally(), is(false)); - - try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) { - LogCaptureAppender.setClassLoggerToTrace(PartitionGroup.class); - assertThat(group.readyToProcess(0L), is(false)); - assertThat( - appender.getEvents(), - hasItem(Matchers.allOf( - Matchers.hasProperty("level", equalTo("TRACE")), - Matchers.hasProperty("message", equalTo("[test] Lag for topic-2 is currently 1, but no data is buffered locally. Waiting to buffer some records.")) - )) - ); - } - } - - @Test - public void shouldIdleAsSpecifiedWhenLagIsZero() { - final PartitionGroup group = new PartitionGroup( - logContext, - mkMap( - mkEntry(partition1, queue1), - mkEntry(partition2, queue2) - ), - getValueSensor(metrics, lastLatenessValue), - enforcedProcessingSensor, - 1L - ); - - final List> list1 = Arrays.asList( - new ConsumerRecord<>("topic", 1, 1L, recordKey, recordValue), - new ConsumerRecord<>("topic", 1, 5L, recordKey, recordValue)); - group.addRawRecords(partition1, list1); - group.addFetchedMetadata(partition2, new ConsumerRecords.Metadata(0L, 0L, 0L)); - - assertThat(group.allPartitionsBufferedLocally(), is(false)); - - try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) { - LogCaptureAppender.setClassLoggerToTrace(PartitionGroup.class); - assertThat(group.readyToProcess(0L), is(false)); - assertThat( - appender.getEvents(), - hasItem(Matchers.allOf( - Matchers.hasProperty("level", equalTo("TRACE")), - Matchers.hasProperty("message", equalTo("[test] Lag for topic-2 is currently 0 and current time is 0. Waiting for new data to be produced for configured idle time 1 (deadline is 1).")) - )) - ); - } - - group.addFetchedMetadata(partition2, new ConsumerRecords.Metadata(0L, 0L, 0L)); - try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) { - LogCaptureAppender.setClassLoggerToTrace(PartitionGroup.class); - assertThat(group.readyToProcess(1L), is(true)); - assertThat( - appender.getEvents(), - hasItem(Matchers.allOf( - Matchers.hasProperty("level", equalTo("INFO")), - Matchers.hasProperty("message", equalTo( - "[test] Continuing to process although some partition timestamps were not buffered locally.\n" + - "\tThere may be out-of-order processing for this task as a result.\n" + - "\tPartitions with local data: [topic-1].\n" + - "\tPartitions we gave up waiting for, with their corresponding deadlines: {topic-2=1}.\n" + - "\tConfigured max.task.idle.ms: 1.\n" + - "\tCurrent wall-clock time: 1." - )) - )) - ); - } - - group.addFetchedMetadata(partition2, new ConsumerRecords.Metadata(0L, 0L, 0L)); - try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) { - LogCaptureAppender.setClassLoggerToTrace(PartitionGroup.class); - assertThat(group.readyToProcess(2L), is(true)); - assertThat( - appender.getEvents(), - hasItem(Matchers.allOf( - Matchers.hasProperty("level", equalTo("INFO")), - Matchers.hasProperty("message", equalTo( - "[test] Continuing to process although some partition timestamps were not buffered locally.\n" + - "\tThere may be out-of-order processing for this task as a result.\n" + - "\tPartitions with local data: [topic-1].\n" + - "\tPartitions we gave up waiting for, with their corresponding deadlines: {topic-2=1}.\n" + - "\tConfigured max.task.idle.ms: 1.\n" + - "\tCurrent wall-clock time: 2." - )) - )) - ); - } - } - private PartitionGroup getBasicGroup() { return new PartitionGroup( - logContext, mkMap( mkEntry(partition1, queue1), mkEntry(partition2, queue2) ), - getValueSensor(metrics, lastLatenessValue), - enforcedProcessingSensor, - maxTaskIdleMs + getValueSensor(metrics, lastLatenessValue) ); } } diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java index ea3fbdd91e1cf..cc871a1541b4a 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java @@ -18,7 +18,6 @@ import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.MockConsumer; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.consumer.OffsetResetStrategy; @@ -40,7 +39,6 @@ import org.apache.kafka.common.serialization.IntegerSerializer; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.serialization.Serializer; -import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.MockTime; import org.apache.kafka.common.utils.Utils; import org.apache.kafka.streams.StreamsConfig; @@ -126,7 +124,6 @@ public class StreamTaskTest { private static final File BASE_DIR = TestUtils.tempDirectory(); private static final long DEFAULT_TIMESTAMP = 1000; - private final LogContext logContext = new LogContext("[test] "); private final String topic1 = "topic1"; private final String topic2 = "topic2"; private final TopicPartition partition1 = new TopicPartition(topic1, 1); @@ -424,9 +421,6 @@ public void shouldProcessInOrder() { assertEquals(asList(101, 102, 103), source1.values); assertEquals(singletonList(201), source2.values); - // tell the task that it doesn't need to wait for more records on partition1 - task.addFetchedMetadata(partition1, new ConsumerRecords.Metadata(0L, 30L, 30L)); - assertTrue(task.process(0L)); assertEquals(1, task.numBuffered()); assertEquals(3, source1.numReceived); @@ -434,8 +428,6 @@ public void shouldProcessInOrder() { assertEquals(asList(101, 102, 103), source1.values); assertEquals(asList(201, 202), source2.values); - // tell the task that it doesn't need to wait for more records on partition1 - task.addFetchedMetadata(partition1, new ConsumerRecords.Metadata(0L, 30L, 30L)); assertTrue(task.process(0L)); assertEquals(0, task.numBuffered()); assertEquals(3, source1.numReceived); @@ -967,9 +959,6 @@ public void shouldPunctuateOnceStreamTimeAfterGap() { assertEquals(3, source2.numReceived); assertTrue(task.maybePunctuateStreamTime()); - // tell the task that it doesn't need to wait for new data on partition1 - task.addFetchedMetadata(partition1, new ConsumerRecords.Metadata(0L, 160L, 160L)); - // st: 161 assertTrue(task.process(0L)); assertEquals(0, task.numBuffered()); @@ -1093,18 +1082,16 @@ public void shouldCommitNextOffsetFromQueueIfAvailable() { @Test public void shouldCommitConsumerPositionIfRecordQueueIsEmpty() { - task = createStatelessTask(createConfig(), StreamsConfig.METRICS_LATEST); + task = createSingleSourceStateless(createConfig(), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); task.completeRestoration(); consumer.addRecord(getConsumerRecordWithOffsetAsTimestamp(partition1, 0L)); consumer.addRecord(getConsumerRecordWithOffsetAsTimestamp(partition1, 1L)); consumer.addRecord(getConsumerRecordWithOffsetAsTimestamp(partition1, 2L)); - consumer.updateEndOffsets(mkMap(mkEntry(partition2, 0L))); consumer.poll(Duration.ZERO); task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(partition1, 0L))); - task.addRecords(partition2, singletonList(getConsumerRecordWithOffsetAsTimestamp(partition2, 0L))); task.process(0L); final Map offsetsAndMetadata = task.prepareCommit(); @@ -1176,6 +1163,107 @@ public void shouldBeProcessableIfAllPartitionsBuffered() { assertTrue(task.process(0L)); } + @Test + public void shouldBeProcessableIfWaitedForTooLong() { + // max idle time is 100ms + task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); + task.initializeIfNeeded(); + task.completeRestoration(); + + final MetricName enforcedProcessMetric = metrics.metricName( + "enforced-processing-total", + "stream-task-metrics", + mkMap(mkEntry("thread-id", Thread.currentThread().getName()), mkEntry("task-id", taskId.toString())) + ); + + assertFalse(task.process(time.milliseconds())); + assertEquals(0.0, metrics.metric(enforcedProcessMetric).metricValue()); + + final byte[] bytes = ByteBuffer.allocate(4).putInt(1).array(); + + task.addRecords(partition1, + asList( + new ConsumerRecord<>(topic1, 1, 0, bytes, bytes), + new ConsumerRecord<>(topic1, 1, 1, bytes, bytes), + new ConsumerRecord<>(topic1, 1, 2, bytes, bytes) + ) + ); + + assertFalse(task.process(time.milliseconds())); + + assertFalse(task.process(time.milliseconds() + 99L)); + + assertTrue(task.process(time.milliseconds() + 100L)); + assertEquals(1.0, metrics.metric(enforcedProcessMetric).metricValue()); + + // once decided to enforce, continue doing that + assertTrue(task.process(time.milliseconds() + 101L)); + assertEquals(2.0, metrics.metric(enforcedProcessMetric).metricValue()); + + task.addRecords(partition2, Collections.singleton(new ConsumerRecord<>(topic2, 1, 0, bytes, bytes))); + + assertTrue(task.process(time.milliseconds() + 130L)); + assertEquals(2.0, metrics.metric(enforcedProcessMetric).metricValue()); + + // one resumed to normal processing, the timer should be reset + + assertFalse(task.process(time.milliseconds() + 150L)); + assertEquals(2.0, metrics.metric(enforcedProcessMetric).metricValue()); + + assertFalse(task.process(time.milliseconds() + 249L)); + assertEquals(2.0, metrics.metric(enforcedProcessMetric).metricValue()); + + assertTrue(task.process(time.milliseconds() + 250L)); + assertEquals(3.0, metrics.metric(enforcedProcessMetric).metricValue()); + } + + @Test + public void shouldNotBeProcessableIfNoDataAvailable() { + task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); + task.initializeIfNeeded(); + task.completeRestoration(); + + final MetricName enforcedProcessMetric = metrics.metricName( + "enforced-processing-total", + "stream-task-metrics", + mkMap(mkEntry("thread-id", Thread.currentThread().getName()), mkEntry("task-id", taskId.toString())) + ); + + assertFalse(task.process(0L)); + assertEquals(0.0, metrics.metric(enforcedProcessMetric).metricValue()); + + final byte[] bytes = ByteBuffer.allocate(4).putInt(1).array(); + + task.addRecords(partition1, Collections.singleton(new ConsumerRecord<>(topic1, 1, 0, bytes, bytes))); + + assertFalse(task.process(time.milliseconds())); + + assertFalse(task.process(time.milliseconds() + 99L)); + + assertTrue(task.process(time.milliseconds() + 100L)); + assertEquals(1.0, metrics.metric(enforcedProcessMetric).metricValue()); + + // once the buffer is drained and no new records coming, the timer should be reset + + assertFalse(task.process(time.milliseconds() + 110L)); + assertEquals(1.0, metrics.metric(enforcedProcessMetric).metricValue()); + + // check that after time is reset, we only falls into enforced processing after the + // whole timeout has elapsed again + task.addRecords(partition1, Collections.singleton(new ConsumerRecord<>(topic1, 1, 0, bytes, bytes))); + + assertFalse(task.process(time.milliseconds() + 150L)); + assertEquals(1.0, metrics.metric(enforcedProcessMetric).metricValue()); + + assertFalse(task.process(time.milliseconds() + 249L)); + assertEquals(1.0, metrics.metric(enforcedProcessMetric).metricValue()); + + assertTrue(task.process(time.milliseconds() + 250L)); + assertEquals(2.0, metrics.metric(enforcedProcessMetric).metricValue()); + } + + + @Test public void shouldPunctuateSystemTimeWhenIntervalElapsed() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); @@ -1550,8 +1638,7 @@ public void shouldReturnOffsetsForRepartitionTopicsForPurging() { time, stateManager, recordCollector, - context, - logContext); + context); task.initializeIfNeeded(); task.completeRestoration(); @@ -1560,7 +1647,6 @@ public void shouldReturnOffsetsForRepartitionTopicsForPurging() { task.addRecords(repartition, singletonList(getConsumerRecordWithOffsetAsTimestamp(repartition, 10L))); assertTrue(task.process(0L)); - task.addFetchedMetadata(partition1, new ConsumerRecords.Metadata(0L, 5L, 5L)); assertTrue(task.process(0L)); task.prepareCommit(); @@ -1691,7 +1777,6 @@ public void shouldCheckpointWhileUpdateSnapshotWithTheConsumedOffsetsForSuspende task.initializeIfNeeded(); task.completeRestoration(); task.addRecords(partition1, singleton(getConsumerRecordWithOffsetAsTimestamp(partition1, 10))); - task.addRecords(partition2, singleton(getConsumerRecordWithOffsetAsTimestamp(partition2, 10))); task.process(100L); assertTrue(task.commitNeeded()); @@ -2021,12 +2106,11 @@ public void shouldThrowIfCleanClosingDirtyTask() { @Test public void shouldThrowIfRecyclingDirtyTask() { - task = createStatelessTask(createConfig(), StreamsConfig.METRICS_LATEST); + task = createSingleSourceStateless(createConfig(), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); task.completeRestoration(); task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(partition1, 0))); - task.addRecords(partition2, singletonList(getConsumerRecordWithOffsetAsTimestamp(partition2, 0))); task.process(0L); assertTrue(task.commitNeeded()); @@ -2117,8 +2201,8 @@ public void shouldThrowTopologyExceptionIfTaskCreatedForUnknownTopic() { time, stateManager, recordCollector, - context, - logContext) + context + ) ); assertThat(exception.getMessage(), equalTo("Invalid topology: " + @@ -2182,8 +2266,7 @@ private StreamTask createOptimizedStatefulTask(final StreamsConfig config, final time, stateManager, recordCollector, - context, - logContext + context ); } @@ -2223,8 +2306,7 @@ public Map committed(final Set clazz) { Logger.getLogger(clazz).setLevel(Level.DEBUG); } - public static void setClassLoggerToTrace(final Class clazz) { - Logger.getLogger(clazz).setLevel(Level.TRACE); - } - public static void unregister(final LogCaptureAppender logCaptureAppender) { Logger.getRootLogger().removeAppender(logCaptureAppender); } diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java index 5f85803812974..20cccf8b2ebf6 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java @@ -447,7 +447,7 @@ private StreamTask createStreamsTask(final StreamsConfig streamsConfig, new MockTime(), stateManager, recordCollector, - context, logContext); + context); } private void mockThread(final boolean initialized) { diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java index 5349d353ad253..8a9007651b8a8 100644 --- a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java +++ b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java @@ -19,7 +19,6 @@ import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerGroupMetadata; import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.MockConsumer; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.consumer.OffsetResetStrategy; @@ -526,21 +525,11 @@ private void setupTask(final StreamsConfig streamsConfig, mockWallClockTime, stateManager, recordCollector, - context, - logContext); + context + ); task.initializeIfNeeded(); task.completeRestoration(); task.processorContext().setRecordContext(null); - - // initialize the task metadata so that all topics have zero lag - for (final Map.Entry entry : startOffsets.entrySet()) { - final ConsumerRecords.Metadata metadata = new ConsumerRecords.Metadata( - mockWallClockTime.milliseconds(), - 0L, - 0L - ); - task.addFetchedMetadata(entry.getKey(), metadata); - } } else { task = null; } @@ -602,11 +591,10 @@ private void enqueueTaskRecord(final String inputTopic, final byte[] key, final byte[] value, final Headers headers) { - final long offset = offsetsByTopicOrPatternPartition.get(topicOrPatternPartition).incrementAndGet() - 1; task.addRecords(topicOrPatternPartition, Collections.singleton(new ConsumerRecord<>( inputTopic, topicOrPatternPartition.partition(), - offset, + offsetsByTopicOrPatternPartition.get(topicOrPatternPartition).incrementAndGet() - 1, timestamp, TimestampType.CREATE_TIME, (long) ConsumerRecord.NULL_CHECKSUM, @@ -616,12 +604,6 @@ private void enqueueTaskRecord(final String inputTopic, value, headers)) ); - final ConsumerRecords.Metadata metadata = new ConsumerRecords.Metadata( - mockWallClockTime.milliseconds(), - offset, - offset - ); - task.addFetchedMetadata(topicOrPatternPartition, metadata); } private void completeAllProcessableWork() { From 413b3af8fed4a7e79c8f44f7385cad8b0c696e05 Mon Sep 17 00:00:00 2001 From: John Roesler Date: Fri, 12 Feb 2021 00:28:17 -0600 Subject: [PATCH 2/2] Revert "KAFKA-10866: Add metadata to ConsumerRecords (#9836)" This reverts commit fdcf8fbf72bee9e672d0790cdbe5539846f7dc8e. --- .../clients/consumer/ConsumerRecords.java | 103 +------------ .../kafka/clients/consumer/KafkaConsumer.java | 7 +- .../kafka/clients/consumer/MockConsumer.java | 17 +-- .../consumer/internals/FetchedRecords.java | 102 ------------- .../clients/consumer/internals/Fetcher.java | 45 +++--- .../consumer/internals/SubscriptionState.java | 12 -- .../clients/consumer/KafkaConsumerTest.java | 137 +----------------- .../consumer/internals/FetcherTest.java | 28 ++-- .../kafka/api/PlaintextConsumerTest.scala | 36 ++--- .../processor/internals/StreamTaskTest.java | 2 +- 10 files changed, 50 insertions(+), 439 deletions(-) delete mode 100644 clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchedRecords.java diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecords.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecords.java index cc5317030bfa4..92390e91907e3 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecords.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecords.java @@ -16,13 +16,11 @@ */ package org.apache.kafka.clients.consumer; -import org.apache.kafka.clients.consumer.internals.FetchedRecords; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.utils.AbstractIterator; import java.util.ArrayList; import java.util.Collections; -import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -34,99 +32,12 @@ * partition returned by a {@link Consumer#poll(java.time.Duration)} operation. */ public class ConsumerRecords implements Iterable> { - public static final ConsumerRecords EMPTY = new ConsumerRecords<>( - Collections.emptyMap(), - Collections.emptyMap() - ); + public static final ConsumerRecords EMPTY = new ConsumerRecords<>(Collections.emptyMap()); private final Map>> records; - private final Map metadata; - public static final class Metadata { - - private final long receivedTimestamp; - private final Long position; - private final Long endOffset; - - public Metadata(final long receivedTimestamp, - final Long position, - final Long endOffset) { - this.receivedTimestamp = receivedTimestamp; - this.position = position; - this.endOffset = endOffset; - } - - /** - * @return The timestamp of the broker response that contained this metadata - */ - public long receivedTimestamp() { - return receivedTimestamp; - } - - /** - * @return The next position the consumer will fetch, or null if the consumer has no position. - */ - public Long position() { - return position; - } - - /** - * @return The lag between the next position to fetch and the current end of the partition, or - * null if the end offset is not known or there is no position. - */ - public Long lag() { - return endOffset == null || position == null ? null : endOffset - position; - } - - /** - * @return The current last offset in the partition. The determination of the "last" offset - * depends on the Consumer's isolation level. Under "read_uncommitted," this is the last successfully - * replicated offset plus one. Under "read_committed," this is the minimum of the last successfully - * replicated offset plus one or the smallest offset of any open transaction. Null if the end offset - * is not known. - */ - public Long endOffset() { - return endOffset; - } - - @Override - public String toString() { - return "Metadata{" + - "receivedTimestamp=" + receivedTimestamp + - ", position=" + position + - ", endOffset=" + endOffset + - '}'; - } - } - - private static Map extractMetadata(final FetchedRecords fetchedRecords) { - final Map metadata = new HashMap<>(); - for (final Map.Entry entry : fetchedRecords.metadata().entrySet()) { - metadata.put( - entry.getKey(), - new Metadata( - entry.getValue().receivedTimestamp(), - entry.getValue().position() == null ? null : entry.getValue().position().offset, - entry.getValue().endOffset() - ) - ); - } - return metadata; - } - - public ConsumerRecords(final Map>> records) { - this.records = records; - this.metadata = new HashMap<>(); - } - - public ConsumerRecords(final Map>> records, - final Map metadata) { + public ConsumerRecords(Map>> records) { this.records = records; - this.metadata = metadata; - } - - ConsumerRecords(final FetchedRecords fetchedRecords) { - this(fetchedRecords.records(), extractMetadata(fetchedRecords)); } /** @@ -142,16 +53,6 @@ public List> records(TopicPartition partition) { return Collections.unmodifiableList(recs); } - /** - * Get the updated metadata returned by the brokers along with this record set. - * May be empty or partial depending on the responses from the broker during this particular poll. - * May also include metadata for additional partitions than the ones for which there are records - * in this {@code ConsumerRecords} object. - */ - public Map metadata() { - return Collections.unmodifiableMap(metadata); - } - /** * Get just the records for the given topic */ diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index e60eebe239c47..b6bebc1717ae0 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -27,7 +27,6 @@ import org.apache.kafka.clients.consumer.internals.ConsumerInterceptors; import org.apache.kafka.clients.consumer.internals.ConsumerMetadata; import org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient; -import org.apache.kafka.clients.consumer.internals.FetchedRecords; import org.apache.kafka.clients.consumer.internals.Fetcher; import org.apache.kafka.clients.consumer.internals.FetcherMetricsRegistry; import org.apache.kafka.clients.consumer.internals.KafkaConsumerMetrics; @@ -1235,7 +1234,7 @@ private ConsumerRecords poll(final Timer timer, final boolean includeMetad } } - final FetchedRecords records = pollForFetches(timer); + final Map>> records = pollForFetches(timer); if (!records.isEmpty()) { // before returning the fetched records, we can send off the next round of fetches // and avoid block waiting for their responses to enable pipelining while the user @@ -1269,12 +1268,12 @@ boolean updateAssignmentMetadataIfNeeded(final Timer timer, final boolean waitFo /** * @throws KafkaException if the rebalance callback throws exception */ - private FetchedRecords pollForFetches(Timer timer) { + private Map>> pollForFetches(Timer timer) { long pollTimeout = coordinator == null ? timer.remainingMs() : Math.min(coordinator.timeToNextPoll(timer.currentTimeMs()), timer.remainingMs()); // if data is available already, return it immediately - final FetchedRecords records = fetcher.fetchedRecords(); + final Map>> records = fetcher.fetchedRecords(); if (!records.isEmpty()) { return records; } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java index 7ddda76e416b0..7bf4c3f16dc94 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java @@ -218,21 +218,7 @@ public synchronized ConsumerRecords poll(final Duration timeout) { } toClear.forEach(p -> this.records.remove(p)); - - final Map metadata = new HashMap<>(); - for (final TopicPartition partition : subscriptions.assignedPartitions()) { - if (subscriptions.hasValidPosition(partition) && endOffsets.containsKey(partition)) { - final SubscriptionState.FetchPosition position = subscriptions.position(partition); - final long offset = position.offset; - final long endOffset = endOffsets.get(partition); - metadata.put( - partition, - new ConsumerRecords.Metadata(System.currentTimeMillis(), offset, endOffset) - ); - } - } - - return new ConsumerRecords<>(results, metadata); + return new ConsumerRecords<>(results); } public synchronized void addRecord(ConsumerRecord record) { @@ -243,7 +229,6 @@ public synchronized void addRecord(ConsumerRecord record) { throw new IllegalStateException("Cannot add records for a partition that is not assigned to the consumer"); List> recs = this.records.computeIfAbsent(tp, k -> new ArrayList<>()); recs.add(record); - endOffsets.compute(tp, (ignore, offset) -> offset == null ? record.offset() : Math.max(offset, record.offset())); } /** diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchedRecords.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchedRecords.java deleted file mode 100644 index d8ef92bd6e48a..0000000000000 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchedRecords.java +++ /dev/null @@ -1,102 +0,0 @@ -/* - * 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.kafka.clients.consumer.internals; - -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.common.TopicPartition; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -public class FetchedRecords { - private final Map>> records; - private final Map metadata; - - public static final class FetchMetadata { - - private final long receivedTimestamp; - private final SubscriptionState.FetchPosition position; - private final Long endOffset; - - public FetchMetadata(final long receivedTimestamp, - final SubscriptionState.FetchPosition position, - final Long endOffset) { - this.receivedTimestamp = receivedTimestamp; - this.position = position; - this.endOffset = endOffset; - } - - public long receivedTimestamp() { - return receivedTimestamp; - } - - public SubscriptionState.FetchPosition position() { - return position; - } - - public Long endOffset() { - return endOffset; - } - - @Override - public String toString() { - return "FetchMetadata{" + - "receivedTimestamp=" + receivedTimestamp + - ", position=" + position + - ", endOffset=" + endOffset + - '}'; - } - } - - public FetchedRecords() { - records = new HashMap<>(); - metadata = new HashMap<>(); - } - - public void addRecords(final TopicPartition topicPartition, final List> records) { - if (this.records.containsKey(topicPartition)) { - // this case shouldn't usually happen because we only send one fetch at a time per partition, - // but it might conceivably happen in some rare cases (such as partition leader changes). - // we have to copy to a new list because the old one may be immutable - final List> currentRecords = this.records.get(topicPartition); - final List> newRecords = new ArrayList<>(records.size() + currentRecords.size()); - newRecords.addAll(currentRecords); - newRecords.addAll(records); - this.records.put(topicPartition, newRecords); - } else { - this.records.put(topicPartition, records); - } - } - - public Map>> records() { - return records; - } - - public void addMetadata(final TopicPartition partition, final FetchMetadata fetchMetadata) { - metadata.put(partition, fetchMetadata); - } - - public Map metadata() { - return metadata; - } - - public boolean isEmpty() { - return records.isEmpty() && metadata.isEmpty(); - } -} diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 01efa762fc9bd..f71d2c45234d5 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -318,7 +318,7 @@ public void onSuccess(ClientResponse resp) { short responseVersion = resp.requestHeader().apiVersion(); completedFetches.add(new CompletedFetch(partition, partitionData, - metricAggregator, batches, fetchOffset, responseVersion, resp.receivedTimeMs())); + metricAggregator, batches, fetchOffset, responseVersion)); } } @@ -597,8 +597,8 @@ private Map beginningOrEndOffset(Collection fetchedRecords() { - FetchedRecords fetched = new FetchedRecords<>(); + public Map>> fetchedRecords() { + Map>> fetched = new HashMap<>(); Queue pausedCompletedFetches = new ArrayDeque<>(); int recordsRemaining = maxPollRecords; @@ -636,28 +636,20 @@ public FetchedRecords fetchedRecords() { } else { List> records = fetchRecords(nextInLineFetch, recordsRemaining); - TopicPartition partition = nextInLineFetch.partition; - - // This can be false when a rebalance happened before fetched records - // are returned to the consumer's poll call - if (subscriptions.isAssigned(partition)) { - - // initializeCompletedFetch, above, has already persisted the metadata from the fetch in the - // SubscriptionState, so we can just read it out, which in particular lets us re-use the logic - // for determining the end offset - final long receivedTimestamp = nextInLineFetch.receivedTimestamp; - final Long endOffset = subscriptions.logEndOffset(partition, isolationLevel); - final FetchPosition fetchPosition = subscriptions.position(partition); - - final FetchedRecords.FetchMetadata metadata = - new FetchedRecords.FetchMetadata(receivedTimestamp, fetchPosition, endOffset); - - fetched.addMetadata(partition, metadata); - - } - if (!records.isEmpty()) { - fetched.addRecords(partition, records); + TopicPartition partition = nextInLineFetch.partition; + List> currentRecords = fetched.get(partition); + if (currentRecords == null) { + fetched.put(partition, records); + } else { + // this case shouldn't usually happen because we only send one fetch at a time per partition, + // but it might conceivably happen in some rare cases (such as partition leader changes). + // we have to copy to a new list because the old one may be immutable + List> newRecords = new ArrayList<>(records.size() + currentRecords.size()); + newRecords.addAll(currentRecords); + newRecords.addAll(records); + fetched.put(partition, newRecords); + } recordsRemaining -= records.size(); } } @@ -1466,7 +1458,6 @@ private class CompletedFetch { private final FetchResponse.PartitionData partitionData; private final FetchResponseMetricAggregator metricAggregator; private final short responseVersion; - private final long receivedTimestamp; private int recordsRead; private int bytesRead; @@ -1485,15 +1476,13 @@ private CompletedFetch(TopicPartition partition, FetchResponseMetricAggregator metricAggregator, Iterator batches, Long fetchOffset, - short responseVersion, - final long receivedTimestamp) { + short responseVersion) { this.partition = partition; this.partitionData = partitionData; this.metricAggregator = metricAggregator; this.batches = batches; this.nextFetchOffset = fetchOffset; this.responseVersion = responseVersion; - this.receivedTimestamp = receivedTimestamp; this.lastEpoch = Optional.empty(); this.abortedProducerIds = new HashSet<>(); this.abortedTransactions = abortedTransactions(partitionData); diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java index b2a5e51aaae8a..30491110a3d78 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java @@ -562,18 +562,6 @@ synchronized void updateLastStableOffset(TopicPartition tp, long lastStableOffse assignedState(tp).lastStableOffset(lastStableOffset); } - synchronized Long logStartOffset(TopicPartition tp) { - return assignedState(tp).logStartOffset; - } - - synchronized Long logEndOffset(TopicPartition tp, IsolationLevel isolationLevel) { - TopicPartitionState topicPartitionState = assignedState(tp); - if (isolationLevel == IsolationLevel.READ_COMMITTED) - return topicPartitionState.lastStableOffset == null ? null : topicPartitionState.lastStableOffset; - else - return topicPartitionState.highWatermark == null ? null : topicPartitionState.highWatermark; - } - /** * Set the preferred read replica with a lease timeout. After this time, the replica will no longer be valid and * {@link #preferredReadReplica(TopicPartition, long)} will return an empty result. diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java index e92e684af0c1c..bf1eed7d5cec1 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java @@ -134,10 +134,6 @@ import static org.apache.kafka.common.requests.FetchMetadata.INVALID_SESSION_ID; import static org.apache.kafka.common.utils.Utils.mkEntry; import static org.apache.kafka.common.utils.Utils.mkMap; -import static org.hamcrest.CoreMatchers.equalTo; -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.greaterThanOrEqualTo; -import static org.hamcrest.Matchers.lessThanOrEqualTo; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; @@ -1056,7 +1052,6 @@ public void fetchResponseWithUnexpectedPartitionIsIgnored() { ConsumerRecords records = consumer.poll(Duration.ZERO); assertEquals(0, records.count()); - assertThat(records.metadata(), equalTo(emptyMap())); consumer.close(Duration.ofMillis(0)); } @@ -1123,9 +1118,7 @@ public void testSubscriptionChangesWithAutoCommitEnabled() { // verify that the fetch occurred as expected assertEquals(11, records.count()); assertEquals(1L, consumer.position(tp0)); - assertEquals(1L, (long) records.metadata().get(tp0).position()); assertEquals(10L, consumer.position(t2p0)); - assertEquals(10L, (long) records.metadata().get(t2p0).position()); // subscription change consumer.subscribe(asList(topic, topic3), getConsumerRebalanceListener(consumer)); @@ -1156,9 +1149,7 @@ public void testSubscriptionChangesWithAutoCommitEnabled() { // verify that the fetch occurred as expected assertEquals(101, records.count()); assertEquals(2L, consumer.position(tp0)); - assertEquals(2L, (long) records.metadata().get(tp0).position()); assertEquals(100L, consumer.position(t3p0)); - assertEquals(100L, (long) records.metadata().get(t3p0).position()); // verify that the offset commits occurred as expected assertTrue(commitReceived.get()); @@ -2052,118 +2043,6 @@ public void testInvalidGroupMetadata() throws InterruptedException { assertThrows(IllegalStateException.class, consumer::groupMetadata); } - @Test - public void testPollMetadata() { - final Time time = new MockTime(); - final SubscriptionState subscription = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST); - final ConsumerMetadata metadata = createMetadata(subscription); - final MockClient client = new MockClient(time, metadata); - - initMetadata(client, singletonMap(topic, 1)); - final ConsumerPartitionAssignor assignor = new RoundRobinAssignor(); - - final KafkaConsumer consumer = - newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); - - consumer.assign(singleton(tp0)); - consumer.seek(tp0, 50L); - - final FetchInfo fetchInfo = new FetchInfo(1L, 99L, 50L, 5); - client.prepareResponse(fetchResponse(singletonMap(tp0, fetchInfo))); - - final ConsumerRecords records = consumer.poll(Duration.ofMillis(1)); - assertEquals(5, records.count()); - assertEquals(55L, consumer.position(tp0)); - - // verify that the consumer computes the correct metadata based on the fetch response - final ConsumerRecords.Metadata actualMetadata = records.metadata().get(tp0); - assertEquals(100L, (long) actualMetadata.endOffset()); - assertEquals(55L, (long) actualMetadata.position()); - assertEquals(45L, (long) actualMetadata.lag()); - consumer.close(Duration.ZERO); - } - - - @Test - public void testPollMetadataWithExtraPartitions() { - final Time time = new MockTime(); - final SubscriptionState subscription = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST); - final ConsumerMetadata metadata = createMetadata(subscription); - final MockClient client = new MockClient(time, metadata); - - initMetadata(client, singletonMap(topic, 2)); - final ConsumerPartitionAssignor assignor = new RoundRobinAssignor(); - - final KafkaConsumer consumer = - newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); - - consumer.assign(asList(tp0, tp1)); - consumer.seek(tp0, 50L); - consumer.seek(tp1, 10L); - - client.prepareResponse( - fetchResponse( - mkMap( - mkEntry(tp0, new FetchInfo(1L, 99L, 50L, 5)), - mkEntry(tp1, new FetchInfo(0L, 29L, 10L, 0)) - ) - ) - ); - - final ConsumerRecords records = consumer.poll(Duration.ofMillis(1)); - assertEquals(5, records.count()); - assertEquals(55L, consumer.position(tp0)); - - assertEquals(5, records.records(tp0).size()); - final ConsumerRecords.Metadata tp0Metadata = records.metadata().get(tp0); - assertEquals(100L, (long) tp0Metadata.endOffset()); - assertEquals(55L, (long) tp0Metadata.position()); - assertEquals(45L, (long) tp0Metadata.lag()); - - // we may get back metadata for other assigned partitions even if we don't get records for them - assertEquals(0, records.records(tp1).size()); - final ConsumerRecords.Metadata tp1Metadata = records.metadata().get(tp1); - assertEquals(30L, (long) tp1Metadata.endOffset()); - assertEquals(10L, (long) tp1Metadata.position()); - assertEquals(20L, (long) tp1Metadata.lag()); - - consumer.close(Duration.ZERO); - } - - @Test - public void testPollMetadataWithNoRecords() { - final Time time = new MockTime(); - final SubscriptionState subscription = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST); - final ConsumerMetadata metadata = createMetadata(subscription); - final MockClient client = new MockClient(time, metadata); - - initMetadata(client, singletonMap(topic, 1)); - final ConsumerPartitionAssignor assignor = new RoundRobinAssignor(); - - final KafkaConsumer consumer = - newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); - - consumer.assign(singleton(tp0)); - consumer.seek(tp0, 50L); - - final FetchInfo fetchInfo = new FetchInfo(1L, 99L, 50L, 0); - client.prepareResponse(fetchResponse(singletonMap(tp0, fetchInfo))); - - final ConsumerRecords records = consumer.poll(Duration.ofMillis(1)); - - // we got no records back ... - assertEquals(0, records.count()); - assertEquals(50L, consumer.position(tp0)); - - // ... but we can still get metadata that was in the fetch response - final ConsumerRecords.Metadata actualMetadata = records.metadata().get(tp0); - assertEquals(100L, (long) actualMetadata.endOffset()); - assertEquals(50L, (long) actualMetadata.position()); - assertEquals(50L, (long) actualMetadata.lag()); - - consumer.close(Duration.ZERO); - } - private KafkaConsumer consumerWithPendingAuthenticationError() { Time time = new MockTime(); SubscriptionState subscription = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST); @@ -2365,8 +2244,6 @@ private FetchResponse fetchResponse(Map fetchResponse(Map( - Errors.NONE, highWatermark, FetchResponse.INVALID_LAST_STABLE_OFFSET, - logStartOffset, null, records)); + Errors.NONE, 0, FetchResponse.INVALID_LAST_STABLE_OFFSET, + 0L, null, records)); } return new FetchResponse<>(Errors.NONE, tpResponses, 0, INVALID_SESSION_ID); } @@ -2504,20 +2381,10 @@ private KafkaConsumer newConsumer(Time time, } private static class FetchInfo { - long logFirstOffset; - long logLastOffset; long offset; int count; FetchInfo(long offset, int count) { - this(0L, offset + count, offset, count); - } - - FetchInfo(long logFirstOffset, long logLastOffset, long offset, int count) { - assertThat(logFirstOffset, lessThanOrEqualTo(offset)); - assertThat(logLastOffset, greaterThanOrEqualTo(offset + count)); - this.logFirstOffset = logFirstOffset; - this.logLastOffset = logLastOffset; this.offset = offset; this.count = count; } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java index c49bb6a700aa3..7af9edec2a33a 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java @@ -521,7 +521,7 @@ public void testParseCorruptedRecord() throws Exception { consumerClient.poll(time.timer(0)); // the first fetchedRecords() should return the first valid message - assertEquals(1, fetcher.fetchedRecords().records().get(tp0).size()); + assertEquals(1, fetcher.fetchedRecords().get(tp0).size()); assertEquals(1, subscriptions.position(tp0).offset); ensureBlockOnRecord(1L); @@ -925,7 +925,7 @@ public void testInFlightFetchOnPausedPartition() { client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0)); consumerClient.poll(time.timer(0)); - assertNull(fetcher.fetchedRecords().records().get(tp0)); + assertNull(fetcher.fetchedRecords().get(tp0)); } @Test @@ -1114,7 +1114,7 @@ public void testFetchNotLeaderOrFollower() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NOT_LEADER_OR_FOLLOWER, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().records().size()); + assertEquals(0, fetcher.fetchedRecords().size()); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds())); } @@ -1127,7 +1127,7 @@ public void testFetchUnknownTopicOrPartition() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.UNKNOWN_TOPIC_OR_PARTITION, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().records().size()); + assertEquals(0, fetcher.fetchedRecords().size()); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds())); } @@ -1141,7 +1141,7 @@ public void testFetchFencedLeaderEpoch() { client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.FENCED_LEADER_EPOCH, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().records().size(), "Should not return any records"); + assertEquals(0, fetcher.fetchedRecords().size(), "Should not return any records"); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds()), "Should have requested metadata update"); } @@ -1155,7 +1155,7 @@ public void testFetchUnknownLeaderEpoch() { client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.UNKNOWN_LEADER_EPOCH, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().records().size(), "Should not return any records"); + assertEquals(0, fetcher.fetchedRecords().size(), "Should not return any records"); assertNotEquals(0L, metadata.timeToNextUpdate(time.milliseconds()), "Should not have requested metadata update"); } @@ -1197,7 +1197,7 @@ public void testFetchOffsetOutOfRange() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.OFFSET_OUT_OF_RANGE, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().records().size()); + assertEquals(0, fetcher.fetchedRecords().size()); assertTrue(subscriptions.isOffsetResetNeeded(tp0)); assertNull(subscriptions.validPosition(tp0)); assertNull(subscriptions.position(tp0)); @@ -1215,7 +1215,7 @@ public void testStaleOutOfRangeError() { client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.OFFSET_OUT_OF_RANGE, 100L, 0)); subscriptions.seek(tp0, 1); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().records().size()); + assertEquals(0, fetcher.fetchedRecords().size()); assertFalse(subscriptions.isOffsetResetNeeded(tp0)); assertEquals(1, subscriptions.position(tp0).offset); } @@ -1233,7 +1233,7 @@ public void testFetchedRecordsAfterSeek() { consumerClient.poll(time.timer(0)); assertFalse(subscriptions.isOffsetResetNeeded(tp0)); subscriptions.seek(tp0, 2); - assertEquals(0, fetcher.fetchedRecords().records().size()); + assertEquals(0, fetcher.fetchedRecords().size()); } @Test @@ -1389,7 +1389,7 @@ public void testSeekBeforeException() { client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(2, fetcher.fetchedRecords().records().get(tp0).size()); + assertEquals(2, fetcher.fetchedRecords().get(tp0).size()); subscriptions.assignFromUser(Utils.mkSet(tp0, tp1)); subscriptions.seekUnvalidated(tp1, new SubscriptionState.FetchPosition(1, Optional.empty(), metadata.currentLeader(tp1))); @@ -1400,11 +1400,11 @@ public void testSeekBeforeException() { FetchResponse.INVALID_LAST_STABLE_OFFSET, FetchResponse.INVALID_LOG_START_OFFSET, Optional.empty(), null, MemoryRecords.EMPTY)); client.prepareResponse(new FetchResponse<>(Errors.NONE, new LinkedHashMap<>(partitions), 0, INVALID_SESSION_ID)); consumerClient.poll(time.timer(0)); - assertEquals(1, fetcher.fetchedRecords().records().get(tp0).size()); + assertEquals(1, fetcher.fetchedRecords().get(tp0).size()); subscriptions.seek(tp1, 10); // Should not throw OffsetOutOfRangeException after the seek - assertEquals(0, fetcher.fetchedRecords().records().size()); + assertEquals(0, fetcher.fetchedRecords().size()); } @Test @@ -1417,7 +1417,7 @@ public void testFetchDisconnected() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0), true); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().records().size()); + assertEquals(0, fetcher.fetchedRecords().size()); // disconnects should have no affect on subscription state assertFalse(subscriptions.isOffsetResetNeeded(tp0)); @@ -4519,7 +4519,7 @@ private MetadataResponse newMetadataResponse(String topic, Errors error) { @SuppressWarnings("unchecked") private Map>> fetchedRecords() { - return (Map) fetcher.fetchedRecords().records(); + return (Map) fetcher.fetchedRecords(); } private void buildFetcher(int maxPollRecords) { diff --git a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala index d0b9084227c49..d4c8492e9a45c 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala @@ -583,7 +583,7 @@ class PlaintextConsumerTest extends BaseConsumerTest { consumer.seekToEnd(List(tp).asJava) assertEquals(totalRecords, consumer.position(tp)) - assertTrue(pollForRecord(consumer, Duration.ofMillis(50)).isEmpty) + assertTrue(consumer.poll(Duration.ofMillis(50)).isEmpty) consumer.seekToBeginning(List(tp).asJava) assertEquals(0L, consumer.position(tp)) @@ -601,7 +601,7 @@ class PlaintextConsumerTest extends BaseConsumerTest { consumer.seekToEnd(List(tp2).asJava) assertEquals(totalRecords, consumer.position(tp2)) - assertTrue(pollForRecord(consumer, Duration.ofMillis(50)).isEmpty) + assertTrue(consumer.poll(Duration.ofMillis(50)).isEmpty) consumer.seekToBeginning(List(tp2).asJava) assertEquals(0L, consumer.position(tp2)) @@ -670,7 +670,7 @@ class PlaintextConsumerTest extends BaseConsumerTest { consumer.pause(partitions) startingTimestamp = System.currentTimeMillis() sendRecords(producer, numRecords = 5, tp, startingTimestamp = startingTimestamp) - assertTrue(pollForRecord(consumer, Duration.ofMillis(100)).isEmpty) + assertTrue(consumer.poll(Duration.ofMillis(100)).isEmpty) consumer.resume(partitions) consumeAndVerifyRecords(consumer = consumer, numRecords = 5, startingOffset = 5, startingTimestamp = startingTimestamp) } @@ -718,8 +718,7 @@ class PlaintextConsumerTest extends BaseConsumerTest { // consuming a record that is too large should succeed since KIP-74 consumer.assign(List(tp).asJava) - val duration = Duration.ofMillis(20000) - val records = pollForRecord(consumer, duration) + val records = consumer.poll(Duration.ofMillis(20000)) assertEquals(1, records.count) val consumerRecord = records.iterator().next() assertEquals(0L, consumerRecord.offset) @@ -751,7 +750,7 @@ class PlaintextConsumerTest extends BaseConsumerTest { // we should only get the small record in the first `poll` consumer.assign(List(tp).asJava) - val records = pollForRecord(consumer, Duration.ofMillis(20000)) + val records = consumer.poll(Duration.ofMillis(20000)) assertEquals(1, records.count) val consumerRecord = records.iterator().next() assertEquals(0L, consumerRecord.offset) @@ -1805,12 +1804,12 @@ class PlaintextConsumerTest extends BaseConsumerTest { consumer3.assign(asList(tp)) consumer3.seek(tp, 1) - val numRecords1 = pollForRecord(consumer1, Duration.ofMillis(5000)).count() + val numRecords1 = consumer1.poll(Duration.ofMillis(5000)).count() assertThrows(classOf[InvalidGroupIdException], () => consumer1.commitSync()) assertThrows(classOf[InvalidGroupIdException], () => consumer2.committed(Set(tp).asJava)) - val numRecords2 = pollForRecord(consumer2, Duration.ofMillis(5000)).count() - val numRecords3 = pollForRecord(consumer3, Duration.ofMillis(5000)).count() + val numRecords2 = consumer2.poll(Duration.ofMillis(5000)).count() + val numRecords3 = consumer3.poll(Duration.ofMillis(5000)).count() consumer1.unsubscribe() consumer2.unsubscribe() @@ -1859,10 +1858,10 @@ class PlaintextConsumerTest extends BaseConsumerTest { consumer1.assign(asList(tp)) consumer2.assign(asList(tp)) - val records1 = pollForRecord(consumer1, Duration.ofMillis(5000)) + val records1 = consumer1.poll(Duration.ofMillis(5000)) consumer1.commitSync() - val records2 = pollForRecord(consumer2, Duration.ofMillis(5000)) + val records2 = consumer2.poll(Duration.ofMillis(5000)) consumer2.commitSync() consumer1.close() @@ -1873,19 +1872,4 @@ class PlaintextConsumerTest extends BaseConsumerTest { assertTrue(records2.count() == 1 && records2.records(tp).asScala.head.offset == 1, "Expected consumer2 to consume one message from offset 1, which is the committed offset of consumer1") } - - /** - * Consumer#poll returns early if there is metadata to return even if there are no records, - * so when we intend to wait for records, we can't just rely on long polling in the Consumer. - */ - private def pollForRecord(consumer: KafkaConsumer[Array[Byte], Array[Byte]], duration: Duration) = { - val deadline = System.currentTimeMillis() + duration.toMillis - var durationRemaining = deadline - System.currentTimeMillis() - var result = consumer.poll(Duration.ofMillis(durationRemaining)) - while (result.count() == 0 && durationRemaining > 0) { - result = consumer.poll(Duration.ofMillis(durationRemaining)) - durationRemaining = deadline - System.currentTimeMillis() - } - result - } } diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java index cc871a1541b4a..f8ed1a5270915 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java @@ -1718,7 +1718,7 @@ public void shouldCheckpointForSuspendedTask() { task.postCommit(true); EasyMock.verify(stateManager); } - + @Test public void shouldNotCheckpointForSuspendedRunningTaskWithSmallProgress() { EasyMock.expect(stateManager.changelogOffsets())