Browse Source
Several issues have come to light since the 2.2.0 release: upon restore, suppress incorrectly set the record metadata using the changelog record, instead of preserving the original metadata restoring a tombstone incorrectly didn't update the buffer size and min-timestamp Reviewers: Guozhang Wang <wangguoz@gmail.com>, Matthias J. Sax <mjsax@apache.org>, Bruno Cadonna <bruno@confluent.io>, Bill Bejeck <bbejeck@gmail.com>pull/6610/head
John Roesler
6 years ago
committed by
Bill Bejeck
12 changed files with 1195 additions and 110 deletions
@ -0,0 +1,604 @@
@@ -0,0 +1,604 @@
|
||||
/* |
||||
* 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.streams.state.internals; |
||||
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord; |
||||
import org.apache.kafka.clients.producer.ProducerRecord; |
||||
import org.apache.kafka.common.header.Header; |
||||
import org.apache.kafka.common.header.internals.RecordHeader; |
||||
import org.apache.kafka.common.header.internals.RecordHeaders; |
||||
import org.apache.kafka.common.record.TimestampType; |
||||
import org.apache.kafka.common.utils.Bytes; |
||||
import org.apache.kafka.common.utils.Utils; |
||||
import org.apache.kafka.streams.KeyValue; |
||||
import org.apache.kafka.streams.StreamsConfig; |
||||
import org.apache.kafka.streams.processor.TaskId; |
||||
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext; |
||||
import org.apache.kafka.streams.processor.internals.RecordBatchingStateRestoreCallback; |
||||
import org.apache.kafka.test.MockInternalProcessorContext; |
||||
import org.apache.kafka.test.MockInternalProcessorContext.MockRecordCollector; |
||||
import org.apache.kafka.test.TestUtils; |
||||
import org.junit.Test; |
||||
import org.junit.runner.RunWith; |
||||
import org.junit.runners.Parameterized; |
||||
|
||||
import java.io.IOException; |
||||
import java.nio.ByteBuffer; |
||||
import java.util.Collection; |
||||
import java.util.LinkedList; |
||||
import java.util.List; |
||||
import java.util.Properties; |
||||
import java.util.Random; |
||||
import java.util.concurrent.atomic.AtomicInteger; |
||||
import java.util.function.Function; |
||||
import java.util.stream.Collectors; |
||||
|
||||
import static java.nio.charset.StandardCharsets.UTF_8; |
||||
import static java.util.Arrays.asList; |
||||
import static java.util.Collections.singletonList; |
||||
import static org.hamcrest.MatcherAssert.assertThat; |
||||
import static org.hamcrest.Matchers.is; |
||||
import static org.junit.Assert.fail; |
||||
|
||||
@RunWith(Parameterized.class) |
||||
public class TimeOrderedKeyValueBufferTest<B extends TimeOrderedKeyValueBuffer> { |
||||
private static final RecordHeaders V_1_CHANGELOG_HEADERS = |
||||
new RecordHeaders(new Header[] {new RecordHeader("v", new byte[] {(byte) 1})}); |
||||
|
||||
private static final String APP_ID = "test-app"; |
||||
private final Function<String, B> bufferSupplier; |
||||
private final String testName; |
||||
|
||||
// As we add more buffer implementations/configurations, we can add them here
|
||||
@Parameterized.Parameters(name = "{index}: test={0}") |
||||
public static Collection<Object[]> parameters() { |
||||
return singletonList( |
||||
new Object[] { |
||||
"in-memory buffer", |
||||
(Function<String, InMemoryTimeOrderedKeyValueBuffer>) name -> |
||||
(InMemoryTimeOrderedKeyValueBuffer) new InMemoryTimeOrderedKeyValueBuffer |
||||
.Builder(name) |
||||
.build() |
||||
} |
||||
); |
||||
} |
||||
|
||||
public TimeOrderedKeyValueBufferTest(final String testName, final Function<String, B> bufferSupplier) { |
||||
this.testName = testName + "_" + new Random().nextInt(Integer.MAX_VALUE); |
||||
this.bufferSupplier = bufferSupplier; |
||||
} |
||||
|
||||
private static MockInternalProcessorContext makeContext() { |
||||
final Properties properties = new Properties(); |
||||
properties.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, APP_ID); |
||||
properties.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, ""); |
||||
|
||||
final TaskId taskId = new TaskId(0, 0); |
||||
|
||||
final MockInternalProcessorContext context = new MockInternalProcessorContext(properties, taskId, TestUtils.tempDirectory()); |
||||
context.setRecordCollector(new MockRecordCollector()); |
||||
|
||||
return context; |
||||
} |
||||
|
||||
|
||||
private static void cleanup(final MockInternalProcessorContext context, final TimeOrderedKeyValueBuffer buffer) { |
||||
try { |
||||
buffer.close(); |
||||
Utils.delete(context.stateDir()); |
||||
} catch (final IOException e) { |
||||
throw new RuntimeException(e); |
||||
} |
||||
} |
||||
|
||||
@Test |
||||
public void shouldInit() { |
||||
final TimeOrderedKeyValueBuffer buffer = bufferSupplier.apply(testName); |
||||
final MockInternalProcessorContext context = makeContext(); |
||||
buffer.init(context, buffer); |
||||
cleanup(context, buffer); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldAcceptData() { |
||||
final TimeOrderedKeyValueBuffer buffer = bufferSupplier.apply(testName); |
||||
final MockInternalProcessorContext context = makeContext(); |
||||
buffer.init(context, buffer); |
||||
putRecord(buffer, context, "2p93nf", 0, "asdf"); |
||||
cleanup(context, buffer); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldRejectNullValues() { |
||||
final TimeOrderedKeyValueBuffer buffer = bufferSupplier.apply(testName); |
||||
final MockInternalProcessorContext context = makeContext(); |
||||
buffer.init(context, buffer); |
||||
try { |
||||
buffer.put(0, getBytes("asdf"), new ContextualRecord( |
||||
null, |
||||
new ProcessorRecordContext(0, 0, 0, "topic") |
||||
)); |
||||
fail("expected an exception"); |
||||
} catch (final NullPointerException expected) { |
||||
// expected
|
||||
} |
||||
cleanup(context, buffer); |
||||
} |
||||
|
||||
private static ContextualRecord getRecord(final String value) { |
||||
return getRecord(value, 0L); |
||||
} |
||||
|
||||
private static ContextualRecord getRecord(final String value, final long timestamp) { |
||||
return new ContextualRecord( |
||||
value.getBytes(UTF_8), |
||||
new ProcessorRecordContext(timestamp, 0, 0, "topic") |
||||
); |
||||
} |
||||
|
||||
private static Bytes getBytes(final String key) { |
||||
return Bytes.wrap(key.getBytes(UTF_8)); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldRemoveData() { |
||||
final TimeOrderedKeyValueBuffer buffer = bufferSupplier.apply(testName); |
||||
final MockInternalProcessorContext context = makeContext(); |
||||
buffer.init(context, buffer); |
||||
putRecord(buffer, context, "qwer", 0, "asdf"); |
||||
assertThat(buffer.numRecords(), is(1)); |
||||
buffer.evictWhile(() -> true, kv -> { }); |
||||
assertThat(buffer.numRecords(), is(0)); |
||||
cleanup(context, buffer); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldRespectEvictionPredicate() { |
||||
final TimeOrderedKeyValueBuffer buffer = bufferSupplier.apply(testName); |
||||
final MockInternalProcessorContext context = makeContext(); |
||||
buffer.init(context, buffer); |
||||
final Bytes firstKey = getBytes("asdf"); |
||||
final ContextualRecord firstRecord = getRecord("eyt"); |
||||
putRecord(0, buffer, context, firstRecord, firstKey); |
||||
putRecord(buffer, context, "rtg", 1, "zxcv"); |
||||
assertThat(buffer.numRecords(), is(2)); |
||||
final List<KeyValue<Bytes, ContextualRecord>> evicted = new LinkedList<>(); |
||||
buffer.evictWhile(() -> buffer.numRecords() > 1, evicted::add); |
||||
assertThat(buffer.numRecords(), is(1)); |
||||
assertThat(evicted, is(singletonList(new KeyValue<>(firstKey, firstRecord)))); |
||||
cleanup(context, buffer); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldTrackCount() { |
||||
final TimeOrderedKeyValueBuffer buffer = bufferSupplier.apply(testName); |
||||
final MockInternalProcessorContext context = makeContext(); |
||||
buffer.init(context, buffer); |
||||
putRecord(buffer, context, "oin", 0, "asdf"); |
||||
assertThat(buffer.numRecords(), is(1)); |
||||
putRecord(buffer, context, "wekjn", 1, "asdf"); |
||||
assertThat(buffer.numRecords(), is(1)); |
||||
putRecord(buffer, context, "24inf", 0, "zxcv"); |
||||
assertThat(buffer.numRecords(), is(2)); |
||||
cleanup(context, buffer); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldTrackSize() { |
||||
final TimeOrderedKeyValueBuffer buffer = bufferSupplier.apply(testName); |
||||
final MockInternalProcessorContext context = makeContext(); |
||||
buffer.init(context, buffer); |
||||
putRecord(buffer, context, "23roni", 0, "asdf"); |
||||
assertThat(buffer.bufferSize(), is(43L)); |
||||
putRecord(buffer, context, "3l", 1, "asdf"); |
||||
assertThat(buffer.bufferSize(), is(39L)); |
||||
putRecord(buffer, context, "qfowin", 0, "zxcv"); |
||||
assertThat(buffer.bufferSize(), is(82L)); |
||||
cleanup(context, buffer); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldTrackMinTimestamp() { |
||||
final TimeOrderedKeyValueBuffer buffer = bufferSupplier.apply(testName); |
||||
final MockInternalProcessorContext context = makeContext(); |
||||
buffer.init(context, buffer); |
||||
putRecord(buffer, context, "2093j", 1, "asdf"); |
||||
assertThat(buffer.minTimestamp(), is(1L)); |
||||
putRecord(buffer, context, "3gon4i", 0, "zxcv"); |
||||
assertThat(buffer.minTimestamp(), is(0L)); |
||||
cleanup(context, buffer); |
||||
} |
||||
|
||||
private static void putRecord(final TimeOrderedKeyValueBuffer buffer, |
||||
final MockInternalProcessorContext context, |
||||
final String value, |
||||
final int time, |
||||
final String key) { |
||||
putRecord(time, buffer, context, getRecord(value), getBytes(key)); |
||||
} |
||||
|
||||
private static void putRecord(final int time, |
||||
final TimeOrderedKeyValueBuffer buffer, |
||||
final MockInternalProcessorContext context, |
||||
final ContextualRecord firstRecord, |
||||
final Bytes firstKey) { |
||||
context.setRecordContext(firstRecord.recordContext()); |
||||
buffer.put(time, firstKey, firstRecord); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldEvictOldestAndUpdateSizeAndCountAndMinTimestamp() { |
||||
final TimeOrderedKeyValueBuffer buffer = bufferSupplier.apply(testName); |
||||
final MockInternalProcessorContext context = makeContext(); |
||||
buffer.init(context, buffer); |
||||
|
||||
putRecord(buffer, context, "o23i4", 1, "zxcv"); |
||||
assertThat(buffer.numRecords(), is(1)); |
||||
assertThat(buffer.bufferSize(), is(42L)); |
||||
assertThat(buffer.minTimestamp(), is(1L)); |
||||
|
||||
putRecord(buffer, context, "3ng", 0, "asdf"); |
||||
assertThat(buffer.numRecords(), is(2)); |
||||
assertThat(buffer.bufferSize(), is(82L)); |
||||
assertThat(buffer.minTimestamp(), is(0L)); |
||||
|
||||
final AtomicInteger callbackCount = new AtomicInteger(0); |
||||
buffer.evictWhile(() -> true, kv -> { |
||||
switch (callbackCount.incrementAndGet()) { |
||||
case 1: { |
||||
assertThat(new String(kv.key.get(), UTF_8), is("asdf")); |
||||
assertThat(buffer.numRecords(), is(2)); |
||||
assertThat(buffer.bufferSize(), is(82L)); |
||||
assertThat(buffer.minTimestamp(), is(0L)); |
||||
break; |
||||
} |
||||
case 2: { |
||||
assertThat(new String(kv.key.get(), UTF_8), is("zxcv")); |
||||
assertThat(buffer.numRecords(), is(1)); |
||||
assertThat(buffer.bufferSize(), is(42L)); |
||||
assertThat(buffer.minTimestamp(), is(1L)); |
||||
break; |
||||
} |
||||
default: { |
||||
fail("too many invocations"); |
||||
break; |
||||
} |
||||
} |
||||
}); |
||||
assertThat(callbackCount.get(), is(2)); |
||||
assertThat(buffer.numRecords(), is(0)); |
||||
assertThat(buffer.bufferSize(), is(0L)); |
||||
assertThat(buffer.minTimestamp(), is(Long.MAX_VALUE)); |
||||
cleanup(context, buffer); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldFlush() { |
||||
final TimeOrderedKeyValueBuffer buffer = bufferSupplier.apply(testName); |
||||
final MockInternalProcessorContext context = makeContext(); |
||||
buffer.init(context, buffer); |
||||
putRecord(2, buffer, context, getRecord("2093j", 0L), getBytes("asdf")); |
||||
putRecord(1, buffer, context, getRecord("3gon4i", 1L), getBytes("zxcv")); |
||||
putRecord(0, buffer, context, getRecord("deadbeef", 2L), getBytes("deleteme")); |
||||
|
||||
// replace "deleteme" with a tombstone
|
||||
buffer.evictWhile(() -> buffer.minTimestamp() < 1, kv -> { }); |
||||
|
||||
// flush everything to the changelog
|
||||
buffer.flush(); |
||||
|
||||
// the buffer should serialize the buffer time and the value as byte[],
|
||||
// which we can't compare for equality using ProducerRecord.
|
||||
// As a workaround, I'm deserializing them and shoving them in a KeyValue, just for ease of testing.
|
||||
|
||||
final List<ProducerRecord<String, KeyValue<Long, ContextualRecord>>> collected = |
||||
((MockRecordCollector) context.recordCollector()) |
||||
.collected() |
||||
.stream() |
||||
.map(pr -> { |
||||
final KeyValue<Long, ContextualRecord> niceValue; |
||||
if (pr.value() == null) { |
||||
niceValue = null; |
||||
} else { |
||||
final byte[] timestampAndValue = pr.value(); |
||||
final ByteBuffer wrap = ByteBuffer.wrap(timestampAndValue); |
||||
final long timestamp = wrap.getLong(); |
||||
final ContextualRecord contextualRecord = ContextualRecord.deserialize(wrap); |
||||
niceValue = new KeyValue<>(timestamp, contextualRecord); |
||||
} |
||||
|
||||
return new ProducerRecord<>(pr.topic(), |
||||
pr.partition(), |
||||
pr.timestamp(), |
||||
new String(pr.key(), UTF_8), |
||||
niceValue, |
||||
pr.headers()); |
||||
}) |
||||
.collect(Collectors.toList()); |
||||
|
||||
assertThat(collected, is(asList( |
||||
new ProducerRecord<>(APP_ID + "-" + testName + "-changelog", |
||||
0, // Producer will assign
|
||||
null, |
||||
"deleteme", |
||||
null, |
||||
new RecordHeaders() |
||||
), |
||||
new ProducerRecord<>(APP_ID + "-" + testName + "-changelog", |
||||
0, |
||||
null, |
||||
"zxcv", |
||||
new KeyValue<>(1L, getRecord("3gon4i", 1)), |
||||
V_1_CHANGELOG_HEADERS |
||||
), |
||||
new ProducerRecord<>(APP_ID + "-" + testName + "-changelog", |
||||
0, |
||||
null, |
||||
"asdf", |
||||
new KeyValue<>(2L, getRecord("2093j", 0)), |
||||
V_1_CHANGELOG_HEADERS |
||||
) |
||||
))); |
||||
|
||||
cleanup(context, buffer); |
||||
} |
||||
|
||||
|
||||
@Test |
||||
public void shouldRestoreOldFormat() { |
||||
final TimeOrderedKeyValueBuffer buffer = bufferSupplier.apply(testName); |
||||
final MockInternalProcessorContext context = makeContext(); |
||||
buffer.init(context, buffer); |
||||
|
||||
final RecordBatchingStateRestoreCallback stateRestoreCallback = |
||||
(RecordBatchingStateRestoreCallback) context.stateRestoreCallback(testName); |
||||
|
||||
context.setRecordContext(new ProcessorRecordContext(0, 0, 0, "")); |
||||
|
||||
stateRestoreCallback.restoreBatch(asList( |
||||
new ConsumerRecord<>("changelog-topic", |
||||
0, |
||||
0, |
||||
0, |
||||
TimestampType.CREATE_TIME, |
||||
-1, |
||||
-1, |
||||
-1, |
||||
"todelete".getBytes(UTF_8), |
||||
ByteBuffer.allocate(Long.BYTES + 6).putLong(0L).put("doomed".getBytes(UTF_8)).array()), |
||||
new ConsumerRecord<>("changelog-topic", |
||||
0, |
||||
1, |
||||
1, |
||||
TimestampType.CREATE_TIME, |
||||
-1, |
||||
-1, |
||||
-1, |
||||
"asdf".getBytes(UTF_8), |
||||
ByteBuffer.allocate(Long.BYTES + 4).putLong(2L).put("qwer".getBytes(UTF_8)).array()), |
||||
new ConsumerRecord<>("changelog-topic", |
||||
0, |
||||
2, |
||||
2, |
||||
TimestampType.CREATE_TIME, |
||||
-1, |
||||
-1, |
||||
-1, |
||||
"zxcv".getBytes(UTF_8), |
||||
ByteBuffer.allocate(Long.BYTES + 5).putLong(1L).put("3o4im".getBytes(UTF_8)).array()) |
||||
)); |
||||
|
||||
assertThat(buffer.numRecords(), is(3)); |
||||
assertThat(buffer.minTimestamp(), is(0L)); |
||||
assertThat(buffer.bufferSize(), is(160L)); |
||||
|
||||
stateRestoreCallback.restoreBatch(singletonList( |
||||
new ConsumerRecord<>("changelog-topic", |
||||
0, |
||||
3, |
||||
3, |
||||
TimestampType.CREATE_TIME, |
||||
-1, |
||||
-1, |
||||
-1, |
||||
"todelete".getBytes(UTF_8), |
||||
null) |
||||
)); |
||||
|
||||
assertThat(buffer.numRecords(), is(2)); |
||||
assertThat(buffer.minTimestamp(), is(1L)); |
||||
assertThat(buffer.bufferSize(), is(103L)); |
||||
|
||||
// flush the buffer into a list in buffer order so we can make assertions about the contents.
|
||||
|
||||
final List<KeyValue<Bytes, ContextualRecord>> evicted = new LinkedList<>(); |
||||
buffer.evictWhile(() -> true, evicted::add); |
||||
|
||||
// Several things to note:
|
||||
// * The buffered records are ordered according to their buffer time (serialized in the value of the changelog)
|
||||
// * The record timestamps are properly restored, and not conflated with the record's buffer time.
|
||||
// * The keys and values are properly restored
|
||||
// * The record topic is set to the changelog topic. This was an oversight in the original implementation,
|
||||
// which is fixed in changelog format v1. But upgraded applications still need to be able to handle the
|
||||
// original format.
|
||||
|
||||
assertThat(evicted, is(asList( |
||||
new KeyValue<>( |
||||
getBytes("zxcv"), |
||||
new ContextualRecord("3o4im".getBytes(UTF_8), |
||||
new ProcessorRecordContext(2, |
||||
2, |
||||
0, |
||||
"changelog-topic", |
||||
new RecordHeaders()))), |
||||
new KeyValue<>( |
||||
getBytes("asdf"), |
||||
new ContextualRecord("qwer".getBytes(UTF_8), |
||||
new ProcessorRecordContext(1, |
||||
1, |
||||
0, |
||||
"changelog-topic", |
||||
new RecordHeaders()))) |
||||
))); |
||||
|
||||
cleanup(context, buffer); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldRestoreNewFormat() { |
||||
final TimeOrderedKeyValueBuffer buffer = bufferSupplier.apply(testName); |
||||
final MockInternalProcessorContext context = makeContext(); |
||||
buffer.init(context, buffer); |
||||
|
||||
final RecordBatchingStateRestoreCallback stateRestoreCallback = |
||||
(RecordBatchingStateRestoreCallback) context.stateRestoreCallback(testName); |
||||
|
||||
context.setRecordContext(new ProcessorRecordContext(0, 0, 0, "")); |
||||
|
||||
final RecordHeaders v1FlagHeaders = new RecordHeaders(new Header[] {new RecordHeader("v", new byte[] {(byte) 1})}); |
||||
|
||||
final byte[] todeleteValue = getRecord("doomed", 0).serialize(); |
||||
final byte[] asdfValue = getRecord("qwer", 1).serialize(); |
||||
final byte[] zxcvValue = getRecord("3o4im", 2).serialize(); |
||||
stateRestoreCallback.restoreBatch(asList( |
||||
new ConsumerRecord<>("changelog-topic", |
||||
0, |
||||
0, |
||||
999, |
||||
TimestampType.CREATE_TIME, |
||||
-1L, |
||||
-1, |
||||
-1, |
||||
"todelete".getBytes(UTF_8), |
||||
ByteBuffer.allocate(Long.BYTES + todeleteValue.length).putLong(0L).put(todeleteValue).array(), |
||||
v1FlagHeaders), |
||||
new ConsumerRecord<>("changelog-topic", |
||||
0, |
||||
1, |
||||
9999, |
||||
TimestampType.CREATE_TIME, |
||||
-1L, |
||||
-1, |
||||
-1, |
||||
"asdf".getBytes(UTF_8), |
||||
ByteBuffer.allocate(Long.BYTES + asdfValue.length).putLong(2L).put(asdfValue).array(), |
||||
v1FlagHeaders), |
||||
new ConsumerRecord<>("changelog-topic", |
||||
0, |
||||
2, |
||||
99, |
||||
TimestampType.CREATE_TIME, |
||||
-1L, |
||||
-1, |
||||
-1, |
||||
"zxcv".getBytes(UTF_8), |
||||
ByteBuffer.allocate(Long.BYTES + zxcvValue.length).putLong(1L).put(zxcvValue).array(), |
||||
v1FlagHeaders) |
||||
)); |
||||
|
||||
assertThat(buffer.numRecords(), is(3)); |
||||
assertThat(buffer.minTimestamp(), is(0L)); |
||||
assertThat(buffer.bufferSize(), is(130L)); |
||||
|
||||
stateRestoreCallback.restoreBatch(singletonList( |
||||
new ConsumerRecord<>("changelog-topic", |
||||
0, |
||||
3, |
||||
3, |
||||
TimestampType.CREATE_TIME, |
||||
-1L, |
||||
-1, |
||||
-1, |
||||
"todelete".getBytes(UTF_8), |
||||
null) |
||||
)); |
||||
|
||||
assertThat(buffer.numRecords(), is(2)); |
||||
assertThat(buffer.minTimestamp(), is(1L)); |
||||
assertThat(buffer.bufferSize(), is(83L)); |
||||
|
||||
// flush the buffer into a list in buffer order so we can make assertions about the contents.
|
||||
|
||||
final List<KeyValue<Bytes, ContextualRecord>> evicted = new LinkedList<>(); |
||||
buffer.evictWhile(() -> true, evicted::add); |
||||
|
||||
// Several things to note:
|
||||
// * The buffered records are ordered according to their buffer time (serialized in the value of the changelog)
|
||||
// * The record timestamps are properly restored, and not conflated with the record's buffer time.
|
||||
// * The keys and values are properly restored
|
||||
// * The record topic is set to the original input topic, *not* the changelog topic
|
||||
// * The record offset preserves the origininal input record's offset, *not* the offset of the changelog record
|
||||
|
||||
|
||||
assertThat(evicted, is(asList( |
||||
new KeyValue<>( |
||||
getBytes("zxcv"), |
||||
new ContextualRecord("3o4im".getBytes(UTF_8), |
||||
new ProcessorRecordContext(2, |
||||
0, |
||||
0, |
||||
"topic", |
||||
null))), |
||||
new KeyValue<>( |
||||
getBytes("asdf"), |
||||
new ContextualRecord("qwer".getBytes(UTF_8), |
||||
new ProcessorRecordContext(1, |
||||
0, |
||||
0, |
||||
"topic", |
||||
null))) |
||||
))); |
||||
|
||||
cleanup(context, buffer); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldNotRestoreUnrecognizedVersionRecord() { |
||||
final TimeOrderedKeyValueBuffer buffer = bufferSupplier.apply(testName); |
||||
final MockInternalProcessorContext context = makeContext(); |
||||
buffer.init(context, buffer); |
||||
|
||||
final RecordBatchingStateRestoreCallback stateRestoreCallback = |
||||
(RecordBatchingStateRestoreCallback) context.stateRestoreCallback(testName); |
||||
|
||||
context.setRecordContext(new ProcessorRecordContext(0, 0, 0, "")); |
||||
|
||||
final RecordHeaders unknownFlagHeaders = new RecordHeaders(new Header[] {new RecordHeader("v", new byte[] {(byte) -1})}); |
||||
|
||||
final byte[] todeleteValue = getRecord("doomed", 0).serialize(); |
||||
try { |
||||
stateRestoreCallback.restoreBatch(singletonList( |
||||
new ConsumerRecord<>("changelog-topic", |
||||
0, |
||||
0, |
||||
999, |
||||
TimestampType.CREATE_TIME, |
||||
-1L, |
||||
-1, |
||||
-1, |
||||
"todelete".getBytes(UTF_8), |
||||
ByteBuffer.allocate(Long.BYTES + todeleteValue.length).putLong(0L).put(todeleteValue).array(), |
||||
unknownFlagHeaders) |
||||
)); |
||||
fail("expected an exception"); |
||||
} catch (final IllegalArgumentException expected) { |
||||
// nothing to do.
|
||||
} finally { |
||||
cleanup(context, buffer); |
||||
} |
||||
} |
||||
} |
||||
Loading…
Reference in new issue