diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SubmittedRecords.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SubmittedRecords.java new file mode 100644 index 0000000000000..472a266f6ea36 --- /dev/null +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SubmittedRecords.java @@ -0,0 +1,297 @@ +/* + * 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.connect.runtime; + +import org.apache.kafka.connect.source.SourceRecord; +import org.apache.kafka.connect.source.SourceTask; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collections; +import java.util.Deque; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.Map; + +/** + * Used to track source records that have been (or are about to be) dispatched to a producer and their accompanying + * source offsets. Records are tracked in the order in which they are submitted, which should match the order they were + * returned from {@link SourceTask#poll()}. The latest-eligible offsets for each source partition can be retrieved via + * {@link #committableOffsets()}, where every record up to and including the record for each returned offset has been + * either {@link SubmittedRecord#ack() acknowledged} or {@link #removeLastOccurrence(SubmittedRecord) removed}. + * Note that this class is not thread-safe, though a {@link SubmittedRecord} can be + * {@link SubmittedRecord#ack() acknowledged} from a different thread. + */ +class SubmittedRecords { + + private static final Logger log = LoggerFactory.getLogger(SubmittedRecords.class); + + // Visible for testing + final Map, Deque> records; + + public SubmittedRecords() { + this.records = new HashMap<>(); + } + + /** + * Enqueue a new source record before dispatching it to a producer. + * The returned {@link SubmittedRecord} should either be {@link SubmittedRecord#ack() acknowledged} in the + * producer callback, or {@link #removeLastOccurrence(SubmittedRecord) removed} if the record could not be successfully + * sent to the producer. + * + * @param record the record about to be dispatched; may not be null but may have a null + * {@link SourceRecord#sourcePartition()} and/or {@link SourceRecord#sourceOffset()} + * @return a {@link SubmittedRecord} that can be either {@link SubmittedRecord#ack() acknowledged} once ack'd by + * the producer, or {@link #removeLastOccurrence removed} if synchronously rejected by the producer + */ + @SuppressWarnings("unchecked") + public SubmittedRecord submit(SourceRecord record) { + return submit((Map) record.sourcePartition(), (Map) record.sourceOffset()); + } + + // Convenience method for testing + SubmittedRecord submit(Map partition, Map offset) { + SubmittedRecord result = new SubmittedRecord(partition, offset); + records.computeIfAbsent(result.partition(), p -> new LinkedList<>()) + .add(result); + return result; + } + + /** + * Remove a source record and do not take it into account any longer when tracking offsets. + * Useful if the record has been synchronously rejected by the producer. + * If multiple instances of the same {@link SubmittedRecord} have been submitted already, only the first one found + * (traversing from the end of the deque backward) will be removed. + * @param record the {@link #submit previously-submitted} record to stop tracking; may not be null + * @return whether an instance of the record was removed + */ + public boolean removeLastOccurrence(SubmittedRecord record) { + Deque deque = records.get(record.partition()); + if (deque == null) { + log.warn("Attempted to remove record from submitted queue for partition {}, but no records with that partition appear to have been submitted", record.partition()); + return false; + } + boolean result = deque.removeLastOccurrence(record); + if (deque.isEmpty()) { + records.remove(record.partition()); + } + if (!result) { + log.warn("Attempted to remove record from submitted queue for partition {}, but the record has not been submitted or has already been removed", record.partition()); + } + return result; + } + + /** + * Clear out any acknowledged records at the head of the deques and return a {@link CommittableOffsets snapshot} of the offsets and offset metadata + * accrued between the last time this method was invoked and now. This snapshot can be {@link CommittableOffsets#updatedWith(CommittableOffsets) combined} + * with an existing snapshot if desired. + * Note that this may take some time to complete if a large number of records has built up, which may occur if a + * Kafka partition is offline and all records targeting that partition go unacknowledged while records targeting + * other partitions continue to be dispatched to the producer and sent successfully + * @return a fresh offset snapshot; never null + */ + public CommittableOffsets committableOffsets() { + Map, Map> offsets = new HashMap<>(); + int totalCommittableMessages = 0; + int totalUncommittableMessages = 0; + int largestDequeSize = 0; + Map largestDequePartition = null; + for (Map.Entry, Deque> entry : records.entrySet()) { + Map partition = entry.getKey(); + Deque queuedRecords = entry.getValue(); + int initialDequeSize = queuedRecords.size(); + if (canCommitHead(queuedRecords)) { + Map offset = committableOffset(queuedRecords); + offsets.put(partition, offset); + } + int uncommittableMessages = queuedRecords.size(); + int committableMessages = initialDequeSize - uncommittableMessages; + totalCommittableMessages += committableMessages; + totalUncommittableMessages += uncommittableMessages; + if (uncommittableMessages > largestDequeSize) { + largestDequeSize = uncommittableMessages; + largestDequePartition = partition; + } + } + // Clear out all empty deques from the map to keep it from growing indefinitely + records.values().removeIf(Deque::isEmpty); + return new CommittableOffsets(offsets, totalCommittableMessages, totalUncommittableMessages, records.size(), largestDequeSize, largestDequePartition); + } + + // Note that this will return null if either there are no committable offsets for the given deque, or the latest + // committable offset is itself null. The caller is responsible for distinguishing between the two cases. + private Map committableOffset(Deque queuedRecords) { + Map result = null; + while (canCommitHead(queuedRecords)) { + result = queuedRecords.poll().offset(); + } + return result; + } + + private boolean canCommitHead(Deque queuedRecords) { + return queuedRecords.peek() != null && queuedRecords.peek().acked(); + } + + static class SubmittedRecord { + private final Map partition; + private final Map offset; + private volatile boolean acked; + + public SubmittedRecord(Map partition, Map offset) { + this.partition = partition; + this.offset = offset; + this.acked = false; + } + + /** + * Acknowledge this record; signals that its offset may be safely committed. + * This is safe to be called from a different thread than what called {@link SubmittedRecords#submit(SourceRecord)}. + */ + public void ack() { + this.acked = true; + } + + private boolean acked() { + return acked; + } + + private Map partition() { + return partition; + } + + private Map offset() { + return offset; + } + } + + /** + * Contains a snapshot of offsets that can be committed for a source task and metadata for that offset commit + * (such as the number of messages for which offsets can and cannot be committed). + */ + static class CommittableOffsets { + + /** + * An "empty" snapshot that contains no offsets to commit and whose metadata contains no committable or uncommitable messages. + */ + public static final CommittableOffsets EMPTY = new CommittableOffsets(Collections.emptyMap(), 0, 0, 0, 0, null); + + private final Map, Map> offsets; + private final int numCommittableMessages; + private final int numUncommittableMessages; + private final int numDeques; + private final int largestDequeSize; + private final Map largestDequePartition; + + CommittableOffsets( + Map, Map> offsets, + int numCommittableMessages, + int numUncommittableMessages, + int numDeques, + int largestDequeSize, + Map largestDequePartition + ) { + this.offsets = offsets != null ? new HashMap<>(offsets) : Collections.emptyMap(); + this.numCommittableMessages = numCommittableMessages; + this.numUncommittableMessages = numUncommittableMessages; + this.numDeques = numDeques; + this.largestDequeSize = largestDequeSize; + this.largestDequePartition = largestDequePartition; + } + + /** + * @return the offsets that can be committed at the time of the snapshot + */ + public Map, Map> offsets() { + return Collections.unmodifiableMap(offsets); + } + + /** + * @return the number of committable messages at the time of the snapshot, where a committable message is both + * acknowledged and not preceded by any unacknowledged messages in the deque for its source partition + */ + public int numCommittableMessages() { + return numCommittableMessages; + } + + /** + * @return the number of uncommittable messages at the time of the snapshot, where an uncommittable message + * is either unacknowledged, or preceded in the deque for its source partition by an unacknowledged message + */ + public int numUncommittableMessages() { + return numUncommittableMessages; + } + + /** + * @return the number of non-empty deques tracking uncommittable messages at the time of the snapshot + */ + public int numDeques() { + return numDeques; + } + + /** + * @return the size of the largest deque at the time of the snapshot + */ + public int largestDequeSize() { + return largestDequeSize; + } + + /** + * Get the partition for the deque with the most uncommitted messages at the time of the snapshot. + * @return the applicable partition, which may be null, or null if there are no uncommitted messages; + * it is the caller's responsibility to distinguish between these two cases via {@link #hasPending()} + */ + public Map largestDequePartition() { + return largestDequePartition; + } + + /** + * @return whether there were any uncommittable messages at the time of the snapshot + */ + public boolean hasPending() { + return numUncommittableMessages > 0; + } + + /** + * @return whether there were any committable or uncommittable messages at the time of the snapshot + */ + public boolean isEmpty() { + return numCommittableMessages == 0 && numUncommittableMessages == 0 && offsets.isEmpty(); + } + + /** + * Create a new snapshot by combining the data for this snapshot with newer data in a more recent snapshot. + * Offsets are combined (giving precedence to the newer snapshot in case of conflict), the total number of + * committable messages is summed across the two snapshots, and the newer snapshot's information on pending + * messages (num deques, largest deque size, etc.) is used. + * @param newerOffsets the newer snapshot to combine with this snapshot + * @return the new offset snapshot containing information from this snapshot and the newer snapshot; never null + */ + public CommittableOffsets updatedWith(CommittableOffsets newerOffsets) { + Map, Map> offsets = new HashMap<>(this.offsets); + offsets.putAll(newerOffsets.offsets); + + return new CommittableOffsets( + offsets, + this.numCommittableMessages + newerOffsets.numCommittableMessages, + newerOffsets.numUncommittableMessages, + newerOffsets.numDeques, + newerOffsets.largestDequeSize, + newerOffsets.largestDequePartition + ); + } + } +} diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java index 7307ec6a4e272..a33821a8dde4c 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java @@ -55,7 +55,6 @@ import org.slf4j.LoggerFactory; import java.time.Duration; -import java.util.IdentityHashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; @@ -66,6 +65,8 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicReference; +import static org.apache.kafka.connect.runtime.SubmittedRecords.SubmittedRecord; +import static org.apache.kafka.connect.runtime.SubmittedRecords.CommittableOffsets; import static org.apache.kafka.connect.runtime.WorkerConfig.TOPIC_TRACKING_ENABLE_CONFIG; /** @@ -94,14 +95,9 @@ class WorkerSourceTask extends WorkerTask { private final TopicCreation topicCreation; private List toSend; - private boolean lastSendFailed; // Whether the last send failed *synchronously*, i.e. never made it into the producer's RecordAccumulator - // Use IdentityHashMap to ensure correctness with duplicate records. This is a HashMap because - // there is no IdentityHashSet. - private IdentityHashMap, ProducerRecord> outstandingMessages; - // A second buffer is used while an offset flush is running - private IdentityHashMap, ProducerRecord> outstandingMessagesBacklog; - private boolean flushing; - private CountDownLatch stopRequestedLatch; + private volatile CommittableOffsets committableOffsets; + private final SubmittedRecords submittedRecords; + private final CountDownLatch stopRequestedLatch; private Map taskConfig; private boolean started = false; @@ -145,10 +141,8 @@ public WorkerSourceTask(ConnectorTaskId id, this.closeExecutor = closeExecutor; this.toSend = null; - this.lastSendFailed = false; - this.outstandingMessages = new IdentityHashMap<>(); - this.outstandingMessagesBacklog = new IdentityHashMap<>(); - this.flushing = false; + this.committableOffsets = CommittableOffsets.EMPTY; + this.submittedRecords = new SubmittedRecords(); this.stopRequestedLatch = new CountDownLatch(1); this.sourceTaskMetricsGroup = new SourceTaskMetricsGroup(id, connectMetrics); this.producerSendException = new AtomicReference<>(); @@ -237,6 +231,8 @@ public void execute() { try { log.info("{} Executing source task", this); while (!isStopping()) { + updateCommittableOffsets(); + if (shouldPause()) { onPause(); if (awaitUnpause()) { @@ -246,7 +242,6 @@ public void execute() { } maybeThrowProducerSendException(); - if (toSend == null) { log.trace("{} Nothing to send to Kafka. Polling source for additional records", this); long start = time.milliseconds(); @@ -255,6 +250,7 @@ public void execute() { recordPollReturned(toSend.size(), time.milliseconds() - start); } } + if (toSend == null) continue; log.trace("{} About to send {} records to Kafka", this, toSend.size()); @@ -268,6 +264,7 @@ public void execute() { // simply resulted in not getting more records but all the existing records should be ok to flush // and commit offsets. Worst case, task.flush() will also throw an exception causing the offset commit // to fail. + updateCommittableOffsets(); commitOffsets(); } } @@ -291,6 +288,13 @@ private void maybeThrowProducerSendException() { } } + private void updateCommittableOffsets() { + CommittableOffsets newOffsets = submittedRecords.committableOffsets(); + synchronized (this) { + this.committableOffsets = this.committableOffsets.updatedWith(newOffsets); + } + } + protected List poll() throws InterruptedException { try { return task.poll(); @@ -352,21 +356,7 @@ private boolean sendRecords() { } log.trace("{} Appending record to the topic {} with key {}, value {}", this, record.topic(), record.key(), record.value()); - // We need this queued first since the callback could happen immediately (even synchronously in some cases). - // Because of this we need to be careful about handling retries -- we always save the previously attempted - // record as part of toSend and need to use a flag to track whether we should actually add it to the outstanding - // messages and update the offsets. - synchronized (this) { - if (!lastSendFailed) { - if (!flushing) { - outstandingMessages.put(producerRecord, producerRecord); - } else { - outstandingMessagesBacklog.put(producerRecord, producerRecord); - } - // Offsets are converted & serialized in the OffsetWriter - offsetWriter.offset(record.sourcePartition(), record.sourceOffset()); - } - } + SubmittedRecord submittedRecord = submittedRecords.submit(record); try { maybeCreateTopic(record.topic()); final String topic = producerRecord.topic(); @@ -378,7 +368,7 @@ private boolean sendRecords() { log.trace("{} Failed record: {}", WorkerSourceTask.this, preTransformRecord); producerSendException.compareAndSet(null, e); } else { - recordSent(producerRecord); + submittedRecord.ack(); counter.completeRecord(); log.trace("{} Wrote record successfully: topic {} partition {} offset {}", WorkerSourceTask.this, @@ -390,12 +380,11 @@ private boolean sendRecords() { } } }); - lastSendFailed = false; } catch (RetriableException | org.apache.kafka.common.errors.RetriableException e) { log.warn("{} Failed to send record to topic '{}' and partition '{}'. Backing off before retrying: ", this, producerRecord.topic(), producerRecord.partition(), e); toSend = toSend.subList(processed, toSend.size()); - lastSendFailed = true; + submittedRecords.removeLastOccurrence(submittedRecord); counter.retryRemaining(); return false; } catch (ConnectException e) { @@ -473,20 +462,6 @@ private void commitTaskRecord(SourceRecord record, RecordMetadata metadata) { } } - private synchronized void recordSent(final ProducerRecord record) { - ProducerRecord removed = outstandingMessages.remove(record); - // While flushing, we may also see callbacks for items in the backlog - if (removed == null && flushing) - removed = outstandingMessagesBacklog.remove(record); - // But if neither one had it, something is very wrong - if (removed == null) { - log.error("{} CRITICAL Saw callback for record from topic {} partition {} that was not present in the outstanding message set", this, record.topic(), record.partition()); - } else if (flushing && outstandingMessages.isEmpty()) { - // flush thread may be waiting on the outstanding messages to clear - this.notifyAll(); - } - } - public boolean commitOffsets() { long commitTimeoutMs = workerConfig.getLong(WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_CONFIG); @@ -495,58 +470,56 @@ public boolean commitOffsets() { long started = time.milliseconds(); long timeout = started + commitTimeoutMs; + CommittableOffsets offsetsToCommit; synchronized (this) { - // First we need to make sure we snapshot everything in exactly the current state. This - // means both the current set of messages we're still waiting to finish, stored in this - // class, which setting flushing = true will handle by storing any new values into a new - // buffer; and the current set of user-specified offsets, stored in the - // OffsetStorageWriter, for which we can use beginFlush() to initiate the snapshot. - flushing = true; - boolean flushStarted = offsetWriter.beginFlush(); - // Still wait for any producer records to flush, even if there aren't any offsets to write - // to persistent storage - - // Next we need to wait for all outstanding messages to finish sending - log.info("{} flushing {} outstanding messages for offset commit", this, outstandingMessages.size()); - while (!outstandingMessages.isEmpty()) { - try { - long timeoutMs = timeout - time.milliseconds(); - // If the task has been cancelled, no more records will be sent from the producer; in that case, if any outstanding messages remain, - // we can stop flushing immediately - if (isCancelled() || timeoutMs <= 0) { - log.error("{} Failed to flush, timed out while waiting for producer to flush outstanding {} messages", this, outstandingMessages.size()); - finishFailedFlush(); - recordCommitFailure(time.milliseconds() - started, null); - return false; - } - this.wait(timeoutMs); - } catch (InterruptedException e) { - // We can get interrupted if we take too long committing when the work thread shutdown is requested, - // requiring a forcible shutdown. Give up since we can't safely commit any offsets, but also need - // to stop immediately - log.error("{} Interrupted while flushing messages, offsets will not be committed", this); - finishFailedFlush(); - recordCommitFailure(time.milliseconds() - started, null); - return false; - } - } + offsetsToCommit = this.committableOffsets; + this.committableOffsets = CommittableOffsets.EMPTY; + } - if (!flushStarted) { - // There was nothing in the offsets to process, but we still waited for the data in the - // buffer to flush. This is useful since this can feed into metrics to monitor, e.g. - // flush time, which can be used for monitoring even if the connector doesn't record any - // offsets. - finishSuccessfulFlush(); - long durationMillis = time.milliseconds() - started; - recordCommitSuccess(durationMillis); - log.debug("{} Finished offset commitOffsets successfully in {} ms", - this, durationMillis); - - commitSourceTask(); - return true; + if (committableOffsets.isEmpty()) { + log.info("{} Either no records were produced by the task since the last offset commit, " + + "or every record has been filtered out by a transformation " + + "or dropped due to transformation or conversion errors.", + this + ); + // We continue with the offset commit process here instead of simply returning immediately + // in order to invoke SourceTask::commit and record metrics for a successful offset commit + } else { + log.info("{} Committing offsets for {} acknowledged messages", this, committableOffsets.numCommittableMessages()); + if (committableOffsets.hasPending()) { + log.debug("{} There are currently {} pending messages spread across {} source partitions whose offsets will not be committed. " + + "The source partition with the most pending messages is {}, with {} pending messages", + this, + committableOffsets.numUncommittableMessages(), + committableOffsets.numDeques(), + committableOffsets.largestDequePartition(), + committableOffsets.largestDequeSize() + ); + } else { + log.debug("{} There are currently no pending messages for this offset commit; " + + "all messages dispatched to the task's producer since the last commit have been acknowledged", + this + ); } } + // Update the offset writer with any new offsets for records that have been acked. + // The offset writer will continue to track all offsets until they are able to be successfully flushed. + // IOW, if the offset writer fails to flush, it keeps those offset for the next attempt, + // though we may update them here with newer offsets for acked records. + offsetsToCommit.offsets().forEach(offsetWriter::offset); + + if (!offsetWriter.beginFlush()) { + // There was nothing in the offsets to process, but we still mark a successful offset commit. + long durationMillis = time.milliseconds() - started; + recordCommitSuccess(durationMillis); + log.debug("{} Finished offset commitOffsets successfully in {} ms", + this, durationMillis); + + commitSourceTask(); + return true; + } + // Now we can actually flush the offsets to user storage. Future flushFuture = offsetWriter.doFlush((error, result) -> { if (error != null) { @@ -558,7 +531,7 @@ public boolean commitOffsets() { // Very rare case: offsets were unserializable and we finished immediately, unable to store // any data if (flushFuture == null) { - finishFailedFlush(); + offsetWriter.cancelFlush(); recordCommitFailure(time.milliseconds() - started, null); return false; } @@ -570,22 +543,21 @@ public boolean commitOffsets() { // could look a little confusing. } catch (InterruptedException e) { log.warn("{} Flush of offsets interrupted, cancelling", this); - finishFailedFlush(); + offsetWriter.cancelFlush(); recordCommitFailure(time.milliseconds() - started, e); return false; } catch (ExecutionException e) { log.error("{} Flush of offsets threw an unexpected exception: ", this, e); - finishFailedFlush(); + offsetWriter.cancelFlush(); recordCommitFailure(time.milliseconds() - started, e); return false; } catch (TimeoutException e) { - log.error("{} Timed out waiting to flush offsets to storage", this); - finishFailedFlush(); + log.error("{} Timed out waiting to flush offsets to storage; will try again on next flush interval with latest offsets", this); + offsetWriter.cancelFlush(); recordCommitFailure(time.milliseconds() - started, null); return false; } - finishSuccessfulFlush(); long durationMillis = time.milliseconds() - started; recordCommitSuccess(durationMillis); log.debug("{} Finished commitOffsets successfully in {} ms", @@ -604,21 +576,6 @@ private void commitSourceTask() { } } - private synchronized void finishFailedFlush() { - offsetWriter.cancelFlush(); - outstandingMessages.putAll(outstandingMessagesBacklog); - outstandingMessagesBacklog.clear(); - flushing = false; - } - - private synchronized void finishSuccessfulFlush() { - // If we were successful, we can just swap instead of replacing items back into the original map - IdentityHashMap, ProducerRecord> temp = outstandingMessages; - outstandingMessages = outstandingMessagesBacklog; - outstandingMessagesBacklog = temp; - flushing = false; - } - @Override public String toString() { return "WorkerSourceTask{" + diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageWriter.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageWriter.java index f8ab6a7f1df55..7766e2cf5d1dc 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageWriter.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageWriter.java @@ -89,9 +89,8 @@ public OffsetStorageWriter(OffsetBackingStore backingStore, * @param partition the partition to store an offset for * @param offset the offset */ - @SuppressWarnings("unchecked") - public synchronized void offset(Map partition, Map offset) { - data.put((Map) partition, (Map) offset); + public synchronized void offset(Map partition, Map offset) { + data.put(partition, offset); } private boolean flushing() { diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java index b0cd0b414f1c8..4743dced429ac 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java @@ -378,8 +378,6 @@ public void testErrorHandlingInSourceTasks() throws Exception { EasyMock.expect(workerSourceTask.commitOffsets()).andReturn(true); - offsetWriter.offset(EasyMock.anyObject(), EasyMock.anyObject()); - EasyMock.expectLastCall().times(2); sourceTask.initialize(EasyMock.anyObject()); EasyMock.expectLastCall(); @@ -444,8 +442,6 @@ public void testErrorHandlingInSourceTasksWthBadConverter() throws Exception { EasyMock.expect(workerSourceTask.commitOffsets()).andReturn(true); - offsetWriter.offset(EasyMock.anyObject(), EasyMock.anyObject()); - EasyMock.expectLastCall().times(2); sourceTask.initialize(EasyMock.anyObject()); EasyMock.expectLastCall(); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SubmittedRecordsTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SubmittedRecordsTest.java new file mode 100644 index 0000000000000..fd339fc165383 --- /dev/null +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SubmittedRecordsTest.java @@ -0,0 +1,289 @@ +/* + * 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.connect.runtime; + +import org.junit.Before; +import org.junit.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.apache.kafka.connect.runtime.SubmittedRecords.SubmittedRecord; +import static org.apache.kafka.connect.runtime.SubmittedRecords.CommittableOffsets; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class SubmittedRecordsTest { + + private static final Map PARTITION1 = Collections.singletonMap("subreddit", "apachekafka"); + private static final Map PARTITION2 = Collections.singletonMap("subreddit", "adifferentvalue"); + private static final Map PARTITION3 = Collections.singletonMap("subreddit", "asdfqweoicus"); + + private AtomicInteger offset; + + SubmittedRecords submittedRecords; + + @Before + public void setup() { + submittedRecords = new SubmittedRecords(); + offset = new AtomicInteger(); + } + + @Test + public void testNoRecords() { + CommittableOffsets committableOffsets = submittedRecords.committableOffsets(); + assertTrue(committableOffsets.isEmpty()); + + committableOffsets = submittedRecords.committableOffsets(); + assertTrue(committableOffsets.isEmpty()); + + committableOffsets = submittedRecords.committableOffsets(); + assertTrue(committableOffsets.isEmpty()); + + assertNoRemainingDeques(); + } + + @Test + public void testNoCommittedRecords() { + for (int i = 0; i < 3; i++) { + for (Map partition : Arrays.asList(PARTITION1, PARTITION2, PARTITION3)) { + submittedRecords.submit(partition, newOffset()); + } + } + + CommittableOffsets committableOffsets = submittedRecords.committableOffsets(); + assertMetadata(committableOffsets, 0, 9, 3, 3, PARTITION1, PARTITION2, PARTITION3); + assertEquals(Collections.emptyMap(), committableOffsets.offsets()); + + committableOffsets = submittedRecords.committableOffsets(); + assertMetadata(committableOffsets, 0, 9, 3, 3, PARTITION1, PARTITION2, PARTITION3); + assertEquals(Collections.emptyMap(), committableOffsets.offsets()); + + committableOffsets = submittedRecords.committableOffsets(); + assertMetadata(committableOffsets, 0, 9, 3, 3, PARTITION1, PARTITION2, PARTITION3); + assertEquals(Collections.emptyMap(), committableOffsets.offsets()); + } + + @Test + public void testSingleAck() { + Map offset = newOffset(); + + SubmittedRecord submittedRecord = submittedRecords.submit(PARTITION1, offset); + CommittableOffsets committableOffsets = submittedRecords.committableOffsets(); + // Record has been submitted but not yet acked; cannot commit offsets for it yet + assertFalse(committableOffsets.isEmpty()); + assertEquals(Collections.emptyMap(), committableOffsets.offsets()); + assertMetadata(committableOffsets, 0, 1, 1, 1, PARTITION1); + assertNoEmptyDeques(); + + submittedRecord.ack(); + committableOffsets = submittedRecords.committableOffsets(); + // Record has been acked; can commit offsets for it + assertFalse(committableOffsets.isEmpty()); + assertEquals(Collections.singletonMap(PARTITION1, offset), committableOffsets.offsets()); + assertMetadataNoPending(committableOffsets, 1); + + // Everything has been ack'd and consumed; make sure that it's been cleaned up to avoid memory leaks + assertNoRemainingDeques(); + + committableOffsets = submittedRecords.committableOffsets(); + // Old offsets should be wiped + assertEquals(Collections.emptyMap(), committableOffsets.offsets()); + assertTrue(committableOffsets.isEmpty()); + } + + @Test + public void testMultipleAcksAcrossMultiplePartitions() { + Map partition1Offset1 = newOffset(); + Map partition1Offset2 = newOffset(); + Map partition2Offset1 = newOffset(); + Map partition2Offset2 = newOffset(); + + SubmittedRecord partition1Record1 = submittedRecords.submit(PARTITION1, partition1Offset1); + SubmittedRecord partition1Record2 = submittedRecords.submit(PARTITION1, partition1Offset2); + SubmittedRecord partition2Record1 = submittedRecords.submit(PARTITION2, partition2Offset1); + SubmittedRecord partition2Record2 = submittedRecords.submit(PARTITION2, partition2Offset2); + + CommittableOffsets committableOffsets = submittedRecords.committableOffsets(); + // No records ack'd yet; can't commit any offsets + assertEquals(Collections.emptyMap(), committableOffsets.offsets()); + assertMetadata(committableOffsets, 0, 4, 2, 2, PARTITION1, PARTITION2); + assertNoEmptyDeques(); + + partition1Record2.ack(); + committableOffsets = submittedRecords.committableOffsets(); + // One record has been ack'd, but a record that comes before it and corresponds to the same source partition hasn't been + assertEquals(Collections.emptyMap(), committableOffsets.offsets()); + assertMetadata(committableOffsets, 0, 4, 2, 2, PARTITION1, PARTITION2); + assertNoEmptyDeques(); + + partition2Record1.ack(); + committableOffsets = submittedRecords.committableOffsets(); + // We can commit the first offset for the second partition + assertEquals(Collections.singletonMap(PARTITION2, partition2Offset1), committableOffsets.offsets()); + assertMetadata(committableOffsets, 1, 3, 2, 2, PARTITION1); + assertNoEmptyDeques(); + + committableOffsets = submittedRecords.committableOffsets(); + // No new offsets to commit + assertEquals(Collections.emptyMap(), committableOffsets.offsets()); + assertMetadata(committableOffsets, 0, 3, 2, 2, PARTITION1); + assertNoEmptyDeques(); + + partition1Record1.ack(); + partition2Record2.ack(); + + committableOffsets = submittedRecords.committableOffsets(); + // We can commit new offsets for both partitions now + Map, Map> expectedOffsets = new HashMap<>(); + expectedOffsets.put(PARTITION1, partition1Offset2); + expectedOffsets.put(PARTITION2, partition2Offset2); + assertEquals(expectedOffsets, committableOffsets.offsets()); + assertMetadataNoPending(committableOffsets, 3); + + // Everything has been ack'd and consumed; make sure that it's been cleaned up to avoid memory leaks + assertNoRemainingDeques(); + + committableOffsets = submittedRecords.committableOffsets(); + // No new offsets to commit + assertTrue(committableOffsets.isEmpty()); + } + + @Test + public void testRemoveLastSubmittedRecord() { + SubmittedRecord submittedRecord = submittedRecords.submit(PARTITION1, newOffset()); + + CommittableOffsets committableOffsets = submittedRecords.committableOffsets(); + assertEquals(Collections.emptyMap(), committableOffsets.offsets()); + assertMetadata(committableOffsets, 0, 1, 1, 1, PARTITION1); + + assertTrue("First attempt to remove record from submitted queue should succeed", submittedRecords.removeLastOccurrence(submittedRecord)); + assertFalse("Attempt to remove already-removed record from submitted queue should fail", submittedRecords.removeLastOccurrence(submittedRecord)); + + committableOffsets = submittedRecords.committableOffsets(); + // Even if SubmittedRecords::remove is broken, we haven't ack'd anything yet, so there should be no committable offsets + assertTrue(committableOffsets.isEmpty()); + + submittedRecord.ack(); + committableOffsets = submittedRecords.committableOffsets(); + // Even though the record has somehow been acknowledged, it should not be counted when collecting committable offsets + assertTrue(committableOffsets.isEmpty()); + } + + @Test + public void testRemoveNotLastSubmittedRecord() { + Map partition1Offset = newOffset(); + Map partition2Offset = newOffset(); + + SubmittedRecord recordToRemove = submittedRecords.submit(PARTITION1, partition1Offset); + SubmittedRecord lastSubmittedRecord = submittedRecords.submit(PARTITION2, partition2Offset); + + CommittableOffsets committableOffsets = submittedRecords.committableOffsets(); + assertMetadata(committableOffsets, 0, 2, 2, 1, PARTITION1, PARTITION2); + assertNoEmptyDeques(); + + assertTrue("First attempt to remove record from submitted queue should succeed", submittedRecords.removeLastOccurrence(recordToRemove)); + + committableOffsets = submittedRecords.committableOffsets(); + // Even if SubmittedRecords::remove is broken, we haven't ack'd anything yet, so there should be no committable offsets + assertEquals(Collections.emptyMap(), committableOffsets.offsets()); + assertMetadata(committableOffsets, 0, 1, 1, 1, PARTITION2); + assertNoEmptyDeques(); + // The only record for this partition has been removed; we shouldn't be tracking a deque for it anymore + assertRemovedDeques(PARTITION1); + + recordToRemove.ack(); + committableOffsets = submittedRecords.committableOffsets(); + // Even though the record has somehow been acknowledged, it should not be counted when collecting committable offsets + assertEquals(Collections.emptyMap(), committableOffsets.offsets()); + assertMetadata(committableOffsets, 0, 1, 1, 1, PARTITION2); + assertNoEmptyDeques(); + + lastSubmittedRecord.ack(); + committableOffsets = submittedRecords.committableOffsets(); + // Now that the last-submitted record has been ack'd, we should be able to commit its offset + assertEquals(Collections.singletonMap(PARTITION2, partition2Offset), committableOffsets.offsets()); + assertMetadata(committableOffsets, 1, 0, 0, 0, (Map) null); + assertFalse(committableOffsets.hasPending()); + + // Everything has been ack'd and consumed; make sure that it's been cleaned up to avoid memory leaks + assertNoRemainingDeques(); + committableOffsets = submittedRecords.committableOffsets(); + assertTrue(committableOffsets.isEmpty()); + } + + @Test + public void testNullPartitionAndOffset() { + SubmittedRecord submittedRecord = submittedRecords.submit(null, null); + CommittableOffsets committableOffsets = submittedRecords.committableOffsets(); + assertMetadata(committableOffsets, 0, 1, 1, 1, (Map) null); + + submittedRecord.ack(); + committableOffsets = submittedRecords.committableOffsets(); + assertEquals(Collections.singletonMap(null, null), committableOffsets.offsets()); + assertMetadataNoPending(committableOffsets, 1); + + assertNoEmptyDeques(); + } + + private void assertNoRemainingDeques() { + assertEquals("Internal records map should be completely empty", Collections.emptyMap(), submittedRecords.records); + } + + @SafeVarargs + private final void assertRemovedDeques(Map... partitions) { + for (Map partition : partitions) { + assertFalse("Deque for partition " + partition + " should have been cleaned up from internal records map", submittedRecords.records.containsKey(partition)); + } + } + + private void assertNoEmptyDeques() { + submittedRecords.records.forEach((partition, deque) -> + assertFalse("Empty deque for partition " + partition + " should have been cleaned up from internal records map", deque.isEmpty()) + ); + } + + private Map newOffset() { + return Collections.singletonMap("timestamp", offset.getAndIncrement()); + } + + private void assertMetadataNoPending(CommittableOffsets committableOffsets, int committableMessages) { + assertEquals(committableMessages, committableOffsets.numCommittableMessages()); + assertFalse(committableOffsets.hasPending()); + } + + @SafeVarargs + @SuppressWarnings("varargs") + private final void assertMetadata( + CommittableOffsets committableOffsets, + int committableMessages, + int uncommittableMessages, + int numDeques, + int largestDequeSize, + Map... largestDequePartitions + ) { + assertEquals(committableMessages, committableOffsets.numCommittableMessages()); + assertEquals(uncommittableMessages, committableOffsets.numUncommittableMessages()); + assertEquals(numDeques, committableOffsets.numDeques()); + assertEquals(largestDequeSize, committableOffsets.largestDequeSize()); + assertTrue(Arrays.asList(largestDequePartitions).contains(committableOffsets.largestDequePartition())); + } +} diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java index 640c1d8259e20..fcd657fb2c142 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java @@ -117,8 +117,8 @@ public class WorkerSourceTaskTest extends ThreadedTest { private static final String TOPIC = "topic"; private static final String OTHER_TOPIC = "other-topic"; - private static final Map PARTITION = Collections.singletonMap("key", "partition".getBytes()); - private static final Map OFFSET = Collections.singletonMap("key", 12); + private static final Map PARTITION = Collections.singletonMap("key", "partition".getBytes()); + private static final Map OFFSET = Collections.singletonMap("key", 12); // Connect-format data private static final Schema KEY_SCHEMA = Schema.INT32_SCHEMA; @@ -286,6 +286,9 @@ public void testPause() throws Exception { sourceTask.stop(); EasyMock.expectLastCall(); + + offsetWriter.offset(PARTITION, OFFSET); + PowerMock.expectLastCall(); expectOffsetFlush(true); statusListener.onShutdown(taskId); @@ -333,6 +336,9 @@ public void testPollsInBackground() throws Exception { sourceTask.stop(); EasyMock.expectLastCall(); + + offsetWriter.offset(PARTITION, OFFSET); + PowerMock.expectLastCall(); expectOffsetFlush(true); statusListener.onShutdown(taskId); @@ -547,6 +553,9 @@ public void testCommit() throws Exception { final CountDownLatch pollLatch = expectPolls(1); expectOffsetFlush(true); + offsetWriter.offset(PARTITION, OFFSET); + PowerMock.expectLastCall().atLeastOnce(); + expectTopicCreation(TOPIC); sourceTask.stop(); @@ -590,6 +599,9 @@ public void testCommitFailure() throws Exception { final CountDownLatch pollLatch = expectPolls(1); expectOffsetFlush(true); + offsetWriter.offset(PARTITION, OFFSET); + PowerMock.expectLastCall().atLeastOnce(); + expectTopicCreation(TOPIC); sourceTask.stop(); @@ -716,25 +728,23 @@ public void testSendRecordsRetries() throws Exception { expectTopicCreation(TOPIC); // First round - expectSendRecordOnce(false); + expectSendRecordOnce(); // Any Producer retriable exception should work here expectSendRecordSyncFailure(new org.apache.kafka.common.errors.TimeoutException("retriable sync failure")); // Second round - expectSendRecordOnce(true); - expectSendRecordOnce(false); + expectSendRecordOnce(); + expectSendRecordOnce(); PowerMock.replayAll(); // Try to send 3, make first pass, second fail. Should save last two Whitebox.setInternalState(workerTask, "toSend", Arrays.asList(record1, record2, record3)); Whitebox.invokeMethod(workerTask, "sendRecords"); - assertEquals(true, Whitebox.getInternalState(workerTask, "lastSendFailed")); assertEquals(Arrays.asList(record2, record3), Whitebox.getInternalState(workerTask, "toSend")); // Next they all succeed Whitebox.invokeMethod(workerTask, "sendRecords"); - assertEquals(false, Whitebox.getInternalState(workerTask, "lastSendFailed")); assertNull(Whitebox.getInternalState(workerTask, "toSend")); PowerMock.verifyAll(); @@ -792,15 +802,14 @@ public void testSendRecordsTaskCommitRecordFail() throws Exception { expectTopicCreation(TOPIC); // Source task commit record failure will not cause the task to abort - expectSendRecordOnce(false); - expectSendRecordTaskCommitRecordFail(false, false); - expectSendRecordOnce(false); + expectSendRecordOnce(); + expectSendRecordTaskCommitRecordFail(false); + expectSendRecordOnce(); PowerMock.replayAll(); Whitebox.setInternalState(workerTask, "toSend", Arrays.asList(record1, record2, record3)); Whitebox.invokeMethod(workerTask, "sendRecords"); - assertEquals(false, Whitebox.getInternalState(workerTask, "lastSendFailed")); assertNull(Whitebox.getInternalState(workerTask, "toSend")); PowerMock.verifyAll(); @@ -930,7 +939,7 @@ public void testHeaders() throws Exception { expectTopicCreation(TOPIC); - Capture> sent = expectSendRecord(TOPIC, true, false, true, true, true, headers); + Capture> sent = expectSendRecord(TOPIC, true, true, true, true, headers); PowerMock.replayAll(); @@ -968,8 +977,8 @@ public void testHeadersWithCustomConverter() throws Exception { expectTopicCreation(TOPIC); - Capture> sentRecordA = expectSendRecord(TOPIC, false, false, true, true, false, null); - Capture> sentRecordB = expectSendRecord(TOPIC, false, false, true, true, false, null); + Capture> sentRecordA = expectSendRecord(TOPIC, false, true, true, false, null); + Capture> sentRecordB = expectSendRecord(TOPIC, false, true, true, false, null); PowerMock.replayAll(); @@ -1009,8 +1018,8 @@ public void testTopicCreateWhenTopicExists() throws Exception { TopicDescription topicDesc = new TopicDescription(TOPIC, false, Collections.singletonList(topicPartitionInfo)); EasyMock.expect(admin.describeTopics(TOPIC)).andReturn(Collections.singletonMap(TOPIC, topicDesc)); - expectSendRecordTaskCommitRecordSucceed(false, false); - expectSendRecordTaskCommitRecordSucceed(false, false); + expectSendRecordTaskCommitRecordSucceed(false); + expectSendRecordTaskCommitRecordSucceed(false); PowerMock.replayAll(); @@ -1037,19 +1046,17 @@ public void testSendRecordsTopicDescribeRetries() throws Exception { // Second round - calls to describe and create succeed expectTopicCreation(TOPIC); // Exactly two records are sent - expectSendRecordTaskCommitRecordSucceed(false, false); - expectSendRecordTaskCommitRecordSucceed(false, false); + expectSendRecordTaskCommitRecordSucceed(false); + expectSendRecordTaskCommitRecordSucceed(false); PowerMock.replayAll(); Whitebox.setInternalState(workerTask, "toSend", Arrays.asList(record1, record2)); Whitebox.invokeMethod(workerTask, "sendRecords"); - assertEquals(true, Whitebox.getInternalState(workerTask, "lastSendFailed")); assertEquals(Arrays.asList(record1, record2), Whitebox.getInternalState(workerTask, "toSend")); // Next they all succeed Whitebox.invokeMethod(workerTask, "sendRecords"); - assertEquals(false, Whitebox.getInternalState(workerTask, "lastSendFailed")); assertNull(Whitebox.getInternalState(workerTask, "toSend")); } @@ -1073,19 +1080,17 @@ public void testSendRecordsTopicCreateRetries() throws Exception { // Second round expectTopicCreation(TOPIC); - expectSendRecordTaskCommitRecordSucceed(false, false); - expectSendRecordTaskCommitRecordSucceed(false, false); + expectSendRecordTaskCommitRecordSucceed(false); + expectSendRecordTaskCommitRecordSucceed(false); PowerMock.replayAll(); Whitebox.setInternalState(workerTask, "toSend", Arrays.asList(record1, record2)); Whitebox.invokeMethod(workerTask, "sendRecords"); - assertEquals(true, Whitebox.getInternalState(workerTask, "lastSendFailed")); assertEquals(Arrays.asList(record1, record2), Whitebox.getInternalState(workerTask, "toSend")); // Next they all succeed Whitebox.invokeMethod(workerTask, "sendRecords"); - assertEquals(false, Whitebox.getInternalState(workerTask, "lastSendFailed")); assertNull(Whitebox.getInternalState(workerTask, "toSend")); } @@ -1105,8 +1110,8 @@ public void testSendRecordsTopicDescribeRetriesMidway() throws Exception { // First round expectPreliminaryCalls(OTHER_TOPIC); expectTopicCreation(TOPIC); - expectSendRecordTaskCommitRecordSucceed(false, false); - expectSendRecordTaskCommitRecordSucceed(false, false); + expectSendRecordTaskCommitRecordSucceed(false); + expectSendRecordTaskCommitRecordSucceed(false); // First call to describe the topic times out EasyMock.expect(admin.describeTopics(OTHER_TOPIC)) @@ -1114,19 +1119,17 @@ public void testSendRecordsTopicDescribeRetriesMidway() throws Exception { // Second round expectTopicCreation(OTHER_TOPIC); - expectSendRecord(OTHER_TOPIC, false, true, true, true, true, emptyHeaders()); + expectSendRecord(OTHER_TOPIC, false, true, true, true, emptyHeaders()); PowerMock.replayAll(); // Try to send 3, make first pass, second fail. Should save last two Whitebox.setInternalState(workerTask, "toSend", Arrays.asList(record1, record2, record3)); Whitebox.invokeMethod(workerTask, "sendRecords"); - assertEquals(true, Whitebox.getInternalState(workerTask, "lastSendFailed")); assertEquals(Arrays.asList(record3), Whitebox.getInternalState(workerTask, "toSend")); // Next they all succeed Whitebox.invokeMethod(workerTask, "sendRecords"); - assertEquals(false, Whitebox.getInternalState(workerTask, "lastSendFailed")); assertNull(Whitebox.getInternalState(workerTask, "toSend")); PowerMock.verifyAll(); @@ -1148,8 +1151,8 @@ public void testSendRecordsTopicCreateRetriesMidway() throws Exception { // First round expectPreliminaryCalls(OTHER_TOPIC); expectTopicCreation(TOPIC); - expectSendRecordTaskCommitRecordSucceed(false, false); - expectSendRecordTaskCommitRecordSucceed(false, false); + expectSendRecordTaskCommitRecordSucceed(false); + expectSendRecordTaskCommitRecordSucceed(false); EasyMock.expect(admin.describeTopics(OTHER_TOPIC)).andReturn(Collections.emptyMap()); // First call to create the topic times out @@ -1159,19 +1162,17 @@ public void testSendRecordsTopicCreateRetriesMidway() throws Exception { // Second round expectTopicCreation(OTHER_TOPIC); - expectSendRecord(OTHER_TOPIC, false, true, true, true, true, emptyHeaders()); + expectSendRecord(OTHER_TOPIC, false, true, true, true, emptyHeaders()); PowerMock.replayAll(); // Try to send 3, make first pass, second fail. Should save last two Whitebox.setInternalState(workerTask, "toSend", Arrays.asList(record1, record2, record3)); Whitebox.invokeMethod(workerTask, "sendRecords"); - assertEquals(true, Whitebox.getInternalState(workerTask, "lastSendFailed")); assertEquals(Arrays.asList(record3), Whitebox.getInternalState(workerTask, "toSend")); // Next they all succeed Whitebox.invokeMethod(workerTask, "sendRecords"); - assertEquals(false, Whitebox.getInternalState(workerTask, "lastSendFailed")); assertNull(Whitebox.getInternalState(workerTask, "toSend")); PowerMock.verifyAll(); @@ -1264,8 +1265,8 @@ public void testTopicCreateSucceedsWhenCreateReturnsExistingTopicFound() throws Capture newTopicCapture = EasyMock.newCapture(); EasyMock.expect(admin.createOrFindTopics(EasyMock.capture(newTopicCapture))).andReturn(foundTopic(TOPIC)); - expectSendRecordTaskCommitRecordSucceed(false, false); - expectSendRecordTaskCommitRecordSucceed(false, false); + expectSendRecordTaskCommitRecordSucceed(false); + expectSendRecordTaskCommitRecordSucceed(false); PowerMock.replayAll(); @@ -1290,8 +1291,8 @@ public void testTopicCreateSucceedsWhenCreateReturnsNewTopicFound() throws Excep Capture newTopicCapture = EasyMock.newCapture(); EasyMock.expect(admin.createOrFindTopics(EasyMock.capture(newTopicCapture))).andReturn(createdTopic(TOPIC)); - expectSendRecordTaskCommitRecordSucceed(false, false); - expectSendRecordTaskCommitRecordSucceed(false, false); + expectSendRecordTaskCommitRecordSucceed(false); + expectSendRecordTaskCommitRecordSucceed(false); PowerMock.replayAll(); @@ -1318,8 +1319,6 @@ private void expectPreliminaryCalls() { private void expectPreliminaryCalls(String topic) { expectConvertHeadersAndKeyValue(topic, true, emptyHeaders()); expectApplyTransformationChain(false); - offsetWriter.offset(PARTITION, OFFSET); - PowerMock.expectLastCall(); } private CountDownLatch expectEmptyPolls(int minimum, final AtomicInteger count) throws InterruptedException { @@ -1363,9 +1362,6 @@ private void expectSendRecordSyncFailure(Throwable error) throws InterruptedExce expectConvertHeadersAndKeyValue(false); expectApplyTransformationChain(false); - offsetWriter.offset(PARTITION, OFFSET); - PowerMock.expectLastCall(); - EasyMock.expect( producer.send(EasyMock.anyObject(ProducerRecord.class), EasyMock.anyObject(org.apache.kafka.clients.producer.Callback.class))) @@ -1373,33 +1369,28 @@ private void expectSendRecordSyncFailure(Throwable error) throws InterruptedExce } private Capture> expectSendRecordAnyTimes() throws InterruptedException { - return expectSendRecordTaskCommitRecordSucceed(true, false); + return expectSendRecordTaskCommitRecordSucceed(true); } - private Capture> expectSendRecordOnce(boolean isRetry) throws InterruptedException { - return expectSendRecordTaskCommitRecordSucceed(false, isRetry); + private Capture> expectSendRecordOnce() throws InterruptedException { + return expectSendRecordTaskCommitRecordSucceed(false); } private Capture> expectSendRecordProducerCallbackFail() throws InterruptedException { - return expectSendRecord(TOPIC, false, false, false, false, true, emptyHeaders()); - } - - private Capture> expectSendRecordTaskCommitRecordSucceed(boolean anyTimes, boolean isRetry) throws InterruptedException { - return expectSendRecord(TOPIC, anyTimes, isRetry, true, true, true, emptyHeaders()); + return expectSendRecord(TOPIC, false, false, false, true, emptyHeaders()); } - private Capture> expectSendRecordTaskCommitRecordFail(boolean anyTimes, boolean isRetry) throws InterruptedException { - return expectSendRecord(TOPIC, anyTimes, isRetry, true, false, true, emptyHeaders()); + private Capture> expectSendRecordTaskCommitRecordSucceed(boolean anyTimes) throws InterruptedException { + return expectSendRecord(TOPIC, anyTimes, true, true, true, emptyHeaders()); } - private Capture> expectSendRecord(boolean anyTimes, boolean isRetry, boolean succeed) throws InterruptedException { - return expectSendRecord(TOPIC, anyTimes, isRetry, succeed, true, true, emptyHeaders()); + private Capture> expectSendRecordTaskCommitRecordFail(boolean anyTimes) throws InterruptedException { + return expectSendRecord(TOPIC, anyTimes, true, false, true, emptyHeaders()); } private Capture> expectSendRecord( String topic, boolean anyTimes, - boolean isRetry, boolean sendSuccess, boolean commitSuccess, boolean isMockedConverters, @@ -1413,16 +1404,7 @@ private Capture> expectSendRecord( Capture> sent = EasyMock.newCapture(); - // 1. Offset data is passed to the offset storage. - if (!isRetry) { - offsetWriter.offset(PARTITION, OFFSET); - if (anyTimes) - PowerMock.expectLastCall().anyTimes(); - else - PowerMock.expectLastCall(); - } - - // 2. Converted data passed to the producer, which will need callbacks invoked for flush to work + // 1. Converted data passed to the producer, which will need callbacks invoked for flush to work IExpectationSetters> expect = EasyMock.expect( producer.send(EasyMock.capture(sent), EasyMock.capture(producerCallbacks))); @@ -1446,7 +1428,7 @@ private Capture> expectSendRecord( expect.andAnswer(expectResponse); if (sendSuccess) { - // 3. As a result of a successful producer send callback, we'll notify the source task of the record commit + // 2. As a result of a successful producer send callback, we'll notify the source task of the record commit expectTaskCommitRecordWithOffset(anyTimes, commitSuccess); expectTaskGetTopic(anyTimes); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/OffsetStorageWriterTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/OffsetStorageWriterTest.java index 38d7a377da885..b442bca86c9ea 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/OffsetStorageWriterTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/OffsetStorageWriterTest.java @@ -47,8 +47,8 @@ public class OffsetStorageWriterTest { private static final String NAMESPACE = "namespace"; // Connect format - any types should be accepted here - private static final Map OFFSET_KEY = Collections.singletonMap("key", "key"); - private static final Map OFFSET_VALUE = Collections.singletonMap("key", 12); + private static final Map OFFSET_KEY = Collections.singletonMap("key", "key"); + private static final Map OFFSET_VALUE = Collections.singletonMap("key", 12); // Serialized private static final byte[] OFFSET_KEY_SERIALIZED = "key-serialized".getBytes(); @@ -229,8 +229,8 @@ public void testCancelAfterAwaitFlush() throws Exception { * ensure tests complete. * @return the captured set of ByteBuffer key-value pairs passed to the storage layer */ - private void expectStore(Map key, byte[] keySerialized, - Map value, byte[] valueSerialized, + private void expectStore(Map key, byte[] keySerialized, + Map value, byte[] valueSerialized, final Callback callback, final boolean fail, final CountDownLatch waitForCompletion) {