Skip to content

KAFKA-3522: Replace RecordConverter with TimestampedBytesStore #6204

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mjsax merged 2 commits into from
Jan 30, 2019
Merged

KAFKA-3522: Replace RecordConverter with TimestampedBytesStore #6204

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f6b402c
KAFKA-3522: Replace RecrodConverter with TimestampedBytesStore
mjsax Jan 26, 2019
a2f34bb
Github comments
mjsax Jan 29, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions ...ms/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,8 @@
import org.apache.kafka.streams.processor.StateRestoreCallback;
import org.apache.kafka.streams.processor.StateRestoreListener;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.state.RecordConverter;
import org.apache.kafka.streams.state.TimestampedBytesStore;
import org.apache.kafka.streams.state.internals.RecordConverter;
import org.apache.kafka.streams.state.internals.WrappedStateStore;
import org.slf4j.Logger;

Expand Down Expand Up @@ -200,7 +201,7 @@ public void register(final StateStore store,
final StateStore stateStore =
store instanceof WrappedStateStore ? ((WrappedStateStore) store).inner() : store;
final RecordConverter recordConverter =
stateStore instanceof RecordConverter ? (RecordConverter) stateStore : new DefaultRecordConverter();
stateStore instanceof TimestampedBytesStore ? RecordConverter.converter() : record -> record;

restoreState(
stateRestoreCallback,
Expand Down
5 changes: 3 additions & 2 deletions ...ams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,9 @@
import org.apache.kafka.streams.processor.StateRestoreCallback;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.state.TimestampedBytesStore;
import org.apache.kafka.streams.state.internals.OffsetCheckpoint;
import org.apache.kafka.streams.state.RecordConverter;
import org.apache.kafka.streams.state.internals.RecordConverter;
import org.apache.kafka.streams.state.internals.WrappedStateStore;
import org.slf4j.Logger;

Expand Down Expand Up @@ -136,7 +137,7 @@ public void register(final StateStore store,
final StateStore stateStore =
store instanceof WrappedStateStore ? ((WrappedStateStore) store).inner() : store;
final RecordConverter recordConverter =
stateStore instanceof RecordConverter ? (RecordConverter) stateStore : new DefaultRecordConverter();
stateStore instanceof TimestampedBytesStore ? RecordConverter.converter() : record -> record;

@guozhangwang guozhangwang Jan 28, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm.. I thought the restorer / state manager would now be agnostic to the convert at all, since the inner store impl is responsible for extending the interface as well as calling its TimestampedBytesStore function internally. Why do we still need an internal RecordConverter class?

@mjsax mjsax Jan 29, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The new static method that is used by the stores, will translate on-disk data from old to new format.

We still need a RecordConverter that translates data from the changelog topic to the new format, because on restore, we put plain <byte[],byte[]> key-value pairs into the store, and the store expects those record to be in the new format.

This make we realize, the the implementation of RecordConverter is actually not correct in this PR -- it should not use the new static method (that inserts a -1 as timestamp), but it need to put the actual record timestamp... Will update the PR accordingly.

@guozhangwang guozhangwang Jan 29, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But can't we make this logic inside the prepareBatch call inside the internal store impl (of course we would require customized users to do so as well) so that the callers do not need to be aware of that?

@mjsax mjsax Jan 29, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, because prepareBatch (or actually restoreAllInternal) takes KeyValue<byte[],byte[] but not a ConsumerRecord. Thus there is not timestamp information we can add to the value.

@vvcephei vvcephei Jan 29, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍 The convertToTimestampedFormat function is purely for converting the already-stored data into the new binary format. It can't actually add any timestamp information, because we don't have it when we retrieve the old-format data. So the function is like (value) -> [-1,value].

In contrast, the state restorer needs to be able to insert the timestamp, so the function is like (value, timestamp) -> [timestamp, value].

Actually, looking at this code again, I see Matthias was right, the RecordConverter erroneously calls through to the convertToTimestampedFormat function. Oops! Good catch!

@vvcephei vvcephei Jan 29, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, I just refreshed, and looked at the new code. It LGTM. Thanks!

@guozhangwang guozhangwang Jan 29, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@mjsax Thanks for the explanation! LGTM.


if (isStandby) {
log.trace("Preparing standby replica of persistent state store {} with changelog topic {}", storeName, topic);
Expand Down
2 changes: 1 addition & 1 deletion streams/src/main/java/org/apache/kafka/streams/processor/internals/StateRestorer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.streams.processor.StateRestoreListener;
import org.apache.kafka.streams.state.RecordConverter;
import org.apache.kafka.streams.state.internals.RecordConverter;

import java.util.ArrayList;
import java.util.Collection;
Expand Down
18 changes: 10 additions & 8 deletions ...sor/internals/DefaultRecordConverter.java → .../streams/state/TimestampedBytesStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,16 +14,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.streams.processor.internals;
package org.apache.kafka.streams.state;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.streams.state.RecordConverter;
import java.nio.ByteBuffer;

public class DefaultRecordConverter implements RecordConverter {
import static org.apache.kafka.clients.consumer.ConsumerRecord.NO_TIMESTAMP;

@Override
public ConsumerRecord<byte[], byte[]> convert(final ConsumerRecord<byte[], byte[]> record) {
return record;
public interface TimestampedBytesStore {
static byte[] convertToTimestampedFormat(final byte[] plainValue) {
return ByteBuffer
.allocate(8 + plainValue.length)
.putLong(NO_TIMESTAMP)
.put(plainValue)
.array();
}

}
42 changes: 29 additions & 13 deletions .../kafka/streams/state/RecordConverter.java → ...eams/state/internals/RecordConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,22 +14,38 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.streams.state;
package org.apache.kafka.streams.state.internals;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.streams.KeyValue;

/**
* {@code RecordConverter} translates a {@link ConsumerRecord} into a {@link KeyValue} pair.
*/
public interface RecordConverter {
import java.nio.ByteBuffer;

/**
* Convert a given record into a key-value pair.
*
* @param record the consumer record
* @return the record as key-value pair
*/
public interface RecordConverter {
ConsumerRecord<byte[], byte[]> convert(final ConsumerRecord<byte[], byte[]> record);

}
@SuppressWarnings("deprecation")
static RecordConverter converter() {
return record -> {
final byte[] rawValue = record.value();
final long timestamp = record.timestamp();
return new ConsumerRecord<>(
record.topic(),
record.partition(),
record.offset(),
timestamp,
record.timestampType(),
record.checksum(),
record.serializedKeySize(),
record.serializedValueSize(),
record.key(),
ByteBuffer
.allocate(8 + rawValue.length)
.putLong(timestamp)
.put(rawValue)
.array(),
record.headers(),
record.leaderEpoch()
);
};
}
}
9 changes: 1 addition & 8 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,9 @@
*/
package org.apache.kafka.streams.state.internals;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.state.RecordConverter;

/**
* A storage engine wrapper for utilities like logging, caching, and metering.
Expand All @@ -40,7 +38,7 @@ public interface WrappedStateStore extends StateStore {
*/
StateStore wrappedStore();

abstract class AbstractStateStore implements WrappedStateStore, RecordConverter {
abstract class AbstractStateStore implements WrappedStateStore {
final StateStore innerState;

protected AbstractStateStore(final StateStore inner) {
Expand Down Expand Up @@ -97,10 +95,5 @@ public StateStore wrappedStore() {
return innerState;
}

@Override
public ConsumerRecord<byte[], byte[]> convert(final ConsumerRecord<byte[], byte[]> record) {
return ((RecordConverter) innerState).convert(record);
}

}
}
29 changes: 10 additions & 19 deletions ...rc/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,8 @@
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateRestoreCallback;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.state.TimestampedBytesStore;
import org.apache.kafka.streams.state.internals.OffsetCheckpoint;
import org.apache.kafka.streams.state.RecordConverter;
import org.apache.kafka.streams.state.internals.WrappedStateStore;
import org.apache.kafka.test.InternalMockProcessorContext;
import org.apache.kafka.test.MockStateRestoreListener;
Expand Down Expand Up @@ -224,7 +224,7 @@ public void shouldThrowStreamsExceptionIfNoPartitionsFoundForStore() {
}

@Test
public void shouldUseDefaultRecordConverterIfStoreDoesNotImplementRecordConverter() {
public void shouldNotConvertValuesIfStoreDoesNotImplementTimestampedBytesStore() {
initializeConsumer(1, 0, t1);

stateManager.initialize();
Expand All @@ -236,7 +236,7 @@ public void shouldUseDefaultRecordConverterIfStoreDoesNotImplementRecordConverte
}

@Test
public void shouldUseDefaultRecordConverterIfInnerStoreDoesNotImplementRecordConverter() {
public void shouldNotConvertValuesIfInnerStoreDoesNotImplementTimestampedBytesStore() {
initializeConsumer(1, 0, t1);

stateManager.initialize();
Expand Down Expand Up @@ -288,20 +288,19 @@ public boolean isOpen() {
}

@Test
public void shouldUseStoreAsRecordConverterIfStoreImplementsRecordConverter() {
public void shouldConvertValuesIfStoreImplementsTimestampedBytesStore() {
initializeConsumer(1, 0, t2);

stateManager.initialize();
stateManager.register(store2, stateRestoreCallback);

final KeyValue<byte[], byte[]> restoredRecord = stateRestoreCallback.restored.get(0);
assertEquals(0, restoredRecord.key.length);
assertEquals(0, restoredRecord.value.length);

assertEquals(3, restoredRecord.key.length);
assertEquals(13, restoredRecord.value.length);
}

@Test
public void shouldUseStoreAsRecordConverterIfInnerStoreImplementsRecordConverter() {
public void shouldConvertValuesIfInnerStoreImplementsTimestampedBytesStore() {
initializeConsumer(1, 0, t2);

stateManager.initialize();
Expand Down Expand Up @@ -348,8 +347,8 @@ public boolean isOpen() {
}, stateRestoreCallback);

final KeyValue<byte[], byte[]> restoredRecord = stateRestoreCallback.restored.get(0);
assertEquals(0, restoredRecord.key.length);
assertEquals(0, restoredRecord.value.length);
assertEquals(3, restoredRecord.key.length);
assertEquals(13, restoredRecord.value.length);
}

@Test
Expand Down Expand Up @@ -827,19 +826,11 @@ public void restore(final byte[] key, final byte[] value) {
}
}



private class ConverterStore<K, V> extends NoOpReadOnlyStore<K, V> implements RecordConverter {

private class ConverterStore<K, V> extends NoOpReadOnlyStore<K, V> implements TimestampedBytesStore {
ConverterStore(final String name,
final boolean rocksdbStore) {
super(name, rocksdbStore);
}

@Override
public ConsumerRecord<byte[], byte[]> convert(final ConsumerRecord<byte[], byte[]> record) {
return new ConsumerRecord<>("", 0, 0L, "".getBytes(), "".getBytes());
}
}

}
19 changes: 6 additions & 13 deletions ...src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,8 @@
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
import org.apache.kafka.streams.state.TimestampedBytesStore;
import org.apache.kafka.streams.state.internals.OffsetCheckpoint;
import org.apache.kafka.streams.state.RecordConverter;
import org.apache.kafka.test.MockBatchingStateRestoreListener;
import org.apache.kafka.test.MockKeyValueStore;
import org.apache.kafka.test.NoOpProcessorContext;
Expand Down Expand Up @@ -153,15 +153,16 @@ public void shouldRestoreStoreWithSinglePutRestoreSpecification() throws Excepti
);
assertThat(persistentStore.keys.size(), is(1));
assertTrue(persistentStore.keys.contains(intKey));
assertEquals(9, persistentStore.values.get(0).length);
} finally {
stateMgr.close(Collections.emptyMap());
}
}

@Test
public void shouldConvertDataOnRestoreIfStoreImplementsRecordConverter() throws Exception {
public void shouldConvertDataOnRestoreIfStoreImplementsTimestampedBytesStore() throws Exception {
final TaskId taskId = new TaskId(0, 2);
final Integer intKey = 2;
final Integer intKey = 1;

final MockKeyValueStore persistentStore = getConverterStore();
final ProcessorStateManager stateMgr = getStandByStateManager(taskId);
Expand All @@ -175,6 +176,7 @@ public void shouldConvertDataOnRestoreIfStoreImplementsRecordConverter() throws
);
assertThat(persistentStore.keys.size(), is(1));
assertTrue(persistentStore.keys.contains(intKey));
assertEquals(17, persistentStore.values.get(0).length);
} finally {
stateMgr.close(Collections.emptyMap());
}
Expand Down Expand Up @@ -797,19 +799,10 @@ private MockKeyValueStore getConverterStore() {
return new ConverterStore("persistentStore", true);
}



private class ConverterStore extends MockKeyValueStore implements RecordConverter {

private class ConverterStore extends MockKeyValueStore implements TimestampedBytesStore {
ConverterStore(final String name,
final boolean persistent) {
super(name, persistent);
}

@Override
public ConsumerRecord<byte[], byte[]> convert(final ConsumerRecord<byte[], byte[]> record) {
return new ConsumerRecord<>("", 0, 0L, new byte[]{0x0, 0x0, 0x0, 0x2}, "".getBytes());
}
}

}
4 changes: 2 additions & 2 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/StateRestorerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ public class StateRestorerTest {
OFFSET_LIMIT,
true,
"storeName",
new DefaultRecordConverter());
record -> record);

@Before
public void setUp() {
Expand Down Expand Up @@ -79,7 +79,7 @@ public void shouldBeCompletedIfOffsetAndOffsetLimitAreZero() {
0,
true,
"storeName",
new DefaultRecordConverter());
record -> record);
assertTrue(restorer.hasCompleted(0, 10));
}

Expand Down
Loading