...
| Code Block |
|---|
| language | java |
|---|
| title | AsyncReadWriteTable |
|---|
| collapse | true |
|---|
|
public interface AsyncReadWriteTable<K, V, U> extends Table {
..
..
/**
* Asynchronously updates an existing record for a given key with the specified update.
*
* @param key the key with which the specified {@code value} is to be associated.
* @param update the update applied to the record associated with a given {@code key}.
* @param args additional arguments
* @throws NullPointerException if the specified {@code key} is {@code null}.
* @return CompletableFuture for the operation
*/
CompletableFuture<Void> updateAsync(K key, U record, Object … args);
/**
* Asynchronously updates the existing records for the given keys with their corresponding updates.
*
* @param updates the key and update mappings.
* @param args additional arguments
* @throws NullPointerException if any of the specified {@code entries} has {@code null} as key.
* @return CompletableFuture for the operation
*/
CompletableFuture<Void> updateAllAsync(List<Entry<K, U>> records, Object … args);
} |
| Code Block |
|---|
| language | java |
|---|
| title | MessageStream |
|---|
| collapse | true |
|---|
|
public interface MessageStream<M> {
/**
* Allows sending update messages in this {@link MessageStream} to a {@link Table} and then propagates this
* {@link MessageStream} to the next chained operator. The type of input message is expected to be {@link KV},
* otherwise a {@link ClassCastException} will be thrown. The value is an UpdateMessage- update and default value.
* Defaults are optional and can be used if the Remote Table integration supports inserting a default through PUT in
* the event an update fails due to an existing record being absent.
* <p>
* Note: The update will be written but may not be flushed to the underlying table before its propagated to the
* chained operators. Whether the message can be read back from the Table in the chained operator depends on whether
* it was flushed and whether the Table offers read after write consistency. Messages retain the original partitioning
* scheme when propagated to next operator.
*
* @param table the table to write messages to
* @param args additional arguments passed to the table
* @param <K> the type of key in the table
* @param <V> the type of record value in the table
* @param <U> the type of update value for the table
* @return this {@link MessageStream}
*/
<K, V, U> MessageStream<KV<K, UpdateMessage<U, V>>> sendUpdateTo(Table<KV<K, V>> table, Object ... args);
} |
Handling First Time Updates
While partial updates are intended to update existing records, there will be certain cases which require support for first-time partial update i.e update to a record which doesn’t exist. To account for such cases, the design needs to have a provision to optionally provide a default record which can be PUT in the absence of an existing record. The update can then be applied on top of the default record. The approach introduces UpdateMessage class which captures the update and an optional default. sendUpdateTo operator which sends updates to a table is designed
| Code Block |
|---|
| language | java |
|---|
| title | UpdateMessage |
|---|
| collapse | true |
|---|
|
/**
* Represents an update and an optional default record to be inserted for a key,
* if the update is applied to a non-existent record.
*
* @param <U> type of the update record
* @param <V> type of the default record
*/
public final class UpdateMessage<U, V> {
private final U update;
@Nullable private final V defaultValue;
public static <U, V> UpdateMessage<U, V> of(U update, @Nullable V defaultValue) {
return new UpdateMessage<>(update, defaultValue);
}
public static <U, V> UpdateMessage<U, V> of(U update) {
return new UpdateMessage<>(update, null);
}
private UpdateMessage(U update, V defaultValue) {
this.update = update;
this.defaultValue = defaultValue;
}
public U getUpdate() {
return update;
}
public V getDefault() {
return defaultValue;
}
} |
