Skip to content

Commit

Permalink
fix(broker-core): fix reprcoessing of deployments
Browse files Browse the repository at this point in the history 
 * add pending deployments to state
 * uses rocks db as store
 * pending deployments are re-distributed at `onOpen` callback
 * push deployments outside of side effects
 * fix key generation on deployments with rocks db
 * introduce new controller abstraction for keys
 * key generator uses controller to store the keys
 * ignore the broker reprocessing tests for now, since they failing due to concurrent reprocessing
  • Loading branch information
@Zelldon
Zelldon committed Aug 27, 2018
1 parent 2f0c4e5 commit 8cfff92
Show file tree
Hide file tree
Showing 15 changed files with 672 additions and 262 deletions.
15 changes: 1 addition & 14 deletions broker-core/src/main/java/io/zeebe/broker/job/processor/JobInstanceStreamProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,6 @@
*/
package io.zeebe.broker.job.processor;

import static io.zeebe.util.StringUtil.getBytes;

import io.zeebe.broker.job.CreditsRequest;
import io.zeebe.broker.job.JobSubscriptionManager;
import io.zeebe.broker.job.data.JobRecord;
Expand All @@ -43,7 +41,6 @@
import io.zeebe.protocol.clientapi.ValueType;
import io.zeebe.protocol.impl.RecordMetadata;
import io.zeebe.protocol.intent.JobIntent;
import java.nio.ByteBuffer;
import java.util.function.Consumer;

public class JobInstanceStreamProcessor implements StreamProcessorLifecycleAware {
Expand All @@ -59,10 +56,6 @@ public class JobInstanceStreamProcessor implements StreamProcessorLifecycleAware

protected int logStreamPartitionId;

private static final byte[] LATEST_JOB_KEY_BUFFER = getBytes("latestJobKey");
private final ByteBuffer dbLongBuffer = ByteBuffer.allocate(Long.BYTES);
private final ByteBuffer dbShortBuffer = ByteBuffer.allocate(Short.BYTES);

public JobInstanceStreamProcessor(JobSubscriptionManager jobSubscriptionManager) {
this.jobSubscriptionManager = jobSubscriptionManager;
}
Expand All @@ -73,7 +66,7 @@ public TypedStreamProcessor createStreamProcessor(TypedStreamEnvironment environ

return environment
.newStreamProcessor()
.keyGenerator(KeyGenerator.createJobKeyGenerator())
.keyGenerator(KeyGenerator.createJobKeyGenerator(stateController))
.onCommand(ValueType.JOB, JobIntent.CREATE, new CreateJobProcessor())
.onCommand(ValueType.JOB, JobIntent.ACTIVATE, new ActivateJobProcessor())
.onCommand(ValueType.JOB, JobIntent.COMPLETE, new CompleteJobProcessor())
Expand All @@ -90,17 +83,11 @@ public StateSnapshotController createSnapshotController(StateStorage storage) {
return new StateSnapshotController(stateController, storage);
}

@Override
public void onOpen(TypedStreamProcessor streamProcessor) {
stateController.recoverLatestJobKey(streamProcessor.getKeyGenerator());
}

private class CreateJobProcessor implements CommandProcessor<JobRecord> {

@Override
public void onCommand(TypedRecord<JobRecord> command, CommandControl commandControl) {
final long jobKey = commandControl.accept(JobIntent.CREATED);
stateController.putLatestJobKey(jobKey);
stateController.putJobState(jobKey, STATE_CREATED);
}
}
Expand Down
74 changes: 11 additions & 63 deletions broker-core/src/main/java/io/zeebe/broker/job/state/JobInstanceStateController.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,60 +17,27 @@
*/
package io.zeebe.broker.job.state;

import static io.zeebe.util.StringUtil.getBytes;
import io.zeebe.broker.util.KeyStateController;
import java.nio.ByteOrder;
import org.agrona.MutableDirectBuffer;
import org.agrona.concurrent.UnsafeBuffer;

import io.zeebe.broker.logstreams.processor.KeyGenerator;
import io.zeebe.logstreams.state.StateController;
import io.zeebe.util.LangUtil;
import java.nio.ByteBuffer;
import org.rocksdb.RocksDBException;

public class JobInstanceStateController extends StateController {
private static final byte[] LATEST_JOB_KEY_BUFFER = getBytes("latestJobKey");
private final ByteBuffer dbLongBuffer = ByteBuffer.allocate(Long.BYTES);
private final ByteBuffer dbShortBuffer = ByteBuffer.allocate(Short.BYTES);

public void recoverLatestJobKey(KeyGenerator keyGenerator) {
ensureIsOpened("recoverLatestJobKey");

if (tryGet(LATEST_JOB_KEY_BUFFER, dbLongBuffer.array())) {
keyGenerator.setKey(dbLongBuffer.getLong(0));
}
}

public void putLatestJobKey(long key) {
ensureIsOpened("putLatestJobKey");

dbLongBuffer.putLong(0, key);

try {
getDb().put(LATEST_JOB_KEY_BUFFER, dbLongBuffer.array());
} catch (RocksDBException e) {
LangUtil.rethrowUnchecked(e);
}
}
public class JobInstanceStateController extends KeyStateController {
private final MutableDirectBuffer dbShortBuffer = new UnsafeBuffer(new byte[Short.BYTES]);

public void putJobState(long key, short state) {
ensureIsOpened("putJobState");

dbLongBuffer.putLong(0, key);
dbShortBuffer.putShort(0, state);

try {
getDb().put(dbLongBuffer.array(), dbShortBuffer.array());
} catch (RocksDBException e) {
LangUtil.rethrowUnchecked(e);
}
dbShortBuffer.putShort(0, state, ByteOrder.LITTLE_ENDIAN);
put(key, dbShortBuffer.byteArray());
}

public short getJobState(long key) {
ensureIsOpened("getJobState");

short state = -1;
dbLongBuffer.putLong(0, key);

if (tryGet(dbLongBuffer.array(), dbShortBuffer.array())) {
state = dbShortBuffer.getShort(0);
if (tryGet(key, dbShortBuffer.byteArray())) {
state = dbShortBuffer.getShort(0, ByteOrder.LITTLE_ENDIAN);
}

return state;
Expand All @@ -79,25 +46,6 @@ public short getJobState(long key) {
public void deleteJobState(long key) {
ensureIsOpened("deleteJobState");

dbLongBuffer.putLong(0, key);

try {
getDb().delete(dbLongBuffer.array());
} catch (RocksDBException e) {
LangUtil.rethrowUnchecked(e);
}
}

private boolean tryGet(final byte[] keyBuffer, final byte[] valueBuffer) {
boolean found = false;

try {
final int bytesRead = getDb().get(keyBuffer, valueBuffer);
found = bytesRead == valueBuffer.length;
} catch (RocksDBException e) {
LangUtil.rethrowUnchecked(e);
}

return found;
delete(key);
}
}
39 changes: 34 additions & 5 deletions broker-core/src/main/java/io/zeebe/broker/logstreams/processor/KeyGenerator.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
*/
package io.zeebe.broker.logstreams.processor;

import io.zeebe.broker.util.KeyStateController;
import io.zeebe.msgpack.UnpackedObject;
import io.zeebe.msgpack.property.LongProperty;

Expand All @@ -34,37 +35,65 @@ public class KeyGenerator extends UnpackedObject {

private final LongProperty nextKey;
private final int stepSize;
private final KeyStateController keyStateController;

public KeyGenerator(long initialValue, int stepSize) {
this(initialValue, stepSize, null);
}

public KeyGenerator(long initialValue, int stepSize, KeyStateController keyStateController) {
nextKey = new LongProperty("nextKey", initialValue);
this.stepSize = stepSize;
declareProperty(nextKey);
this.keyStateController = keyStateController;
init();
}

private void init() {
if (keyStateController != null) {
keyStateController.addOnOpenCallback(
() -> {
final long latestKey = keyStateController.getLatestKey();
if (latestKey > 0) {
setKey(latestKey);
}
});
}
}

public long nextKey() {
final long key = nextKey.getValue();
nextKey.setValue(key + stepSize);
putLatestKeyIntoController(key);
return key;
}

public void setKey(long key) {
nextKey.setValue(key + stepSize);
final long nextKey = key + stepSize;
this.nextKey.setValue(nextKey);
putLatestKeyIntoController(key);
}

private void putLatestKeyIntoController(long key) {
if (keyStateController != null) {
keyStateController.putLatestKey(key);
}
}

public static KeyGenerator createWorkflowInstanceKeyGenerator() {
return new KeyGenerator(WF_OFFSET, STEP_SIZE);
}

public static KeyGenerator createJobKeyGenerator() {
return new KeyGenerator(JOB_OFFSET, STEP_SIZE);
public static KeyGenerator createJobKeyGenerator(KeyStateController keyStateController) {
return new KeyGenerator(JOB_OFFSET, STEP_SIZE, keyStateController);
}

public static KeyGenerator createIncidentKeyGenerator() {
return new KeyGenerator(INCIDENT_OFFSET, STEP_SIZE);
}

public static KeyGenerator createDeploymentKeyGenerator() {
return new KeyGenerator(DEPLOYMENT_OFFSET, STEP_SIZE);
public static KeyGenerator createDeploymentKeyGenerator(KeyStateController keyStateController) {
return new KeyGenerator(DEPLOYMENT_OFFSET, STEP_SIZE, keyStateController);
}

public static KeyGenerator createTopicKeyGenerator() {
Expand Down
42 changes: 30 additions & 12 deletions .../io/zeebe/broker/system/workflow/repository/processor/DeploymentCreateEventProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import io.zeebe.broker.logstreams.processor.TypedStreamProcessor;
import io.zeebe.broker.logstreams.processor.TypedStreamWriter;
import io.zeebe.broker.system.workflow.repository.data.DeploymentRecord;
import io.zeebe.broker.system.workflow.repository.processor.state.DeploymentsStateController;
import io.zeebe.broker.system.workflow.repository.processor.state.WorkflowRepositoryIndex;
import io.zeebe.logstreams.log.LogStreamWriterImpl;
import io.zeebe.logstreams.processor.EventLifecycleContext;
Expand All @@ -46,6 +47,7 @@ public class DeploymentCreateEventProcessor implements TypedRecordProcessor<Depl
private final DeploymentTransformer deploymentTransformer;
private final LogStreamWriterImpl logStreamWriter;
private final ClientTransport managementApi;
private final DeploymentsStateController deploymentsStateController;

private ActorControl actor;
private TopologyPartitionListenerImpl partitionListener;
Expand All @@ -55,9 +57,11 @@ public class DeploymentCreateEventProcessor implements TypedRecordProcessor<Depl
public DeploymentCreateEventProcessor(
TopologyManager topologyManager,
WorkflowRepositoryIndex index,
DeploymentsStateController deploymentsStateController,
ClientTransport managementClient,
LogStreamWriterImpl logStreamWriter) {
deploymentTransformer = new DeploymentTransformer(index);
this.deploymentsStateController = deploymentsStateController;
this.topologyManager = topologyManager;
managementApi = managementClient;
this.logStreamWriter = logStreamWriter;
Expand All @@ -71,7 +75,22 @@ public void onOpen(TypedStreamProcessor streamProcessor) {
partitionListener = new TopologyPartitionListenerImpl(streamProcessor.getActor());
topologyManager.addTopologyPartitionListener(partitionListener);

deploymentDistributor = new DeploymentDistributor(managementApi, partitionListener, actor);
deploymentDistributor =
new DeploymentDistributor(
managementApi, partitionListener, deploymentsStateController, actor);

actor.submit(this::reprocessPendingDeployments);
}

private void reprocessPendingDeployments() {
deploymentsStateController.foreach(
((key, pendingDeploymentDistribution) -> {
final ExpandableArrayBuffer buffer = new ExpandableArrayBuffer();
final DirectBuffer deployment = pendingDeploymentDistribution.getDeployment();
buffer.putBytes(0, deployment, 0, deployment.capacity());

deploymentCreation(key, pendingDeploymentDistribution.getSourcePosition(), buffer);
}));
}

@Override
Expand Down Expand Up @@ -105,19 +124,18 @@ private void processValidDeployment(
DeploymentRecord deploymentEvent) {
final long key = streamWriter.getKeyGenerator().nextKey();

sideEffect.accept(
() -> {
final ExpandableArrayBuffer buffer = new ExpandableArrayBuffer();
deploymentEvent.write(buffer, 0);
final ActorFuture<Void> pushDeployment =
deploymentDistributor.pushDeployment(key, event.getPosition(), buffer);
final ExpandableArrayBuffer buffer = new ExpandableArrayBuffer();
deploymentEvent.write(buffer, 0);
deploymentCreation(key, event.getPosition(), buffer);

actor.runOnCompletion(
pushDeployment, (aVoid, throwable) -> writeDeploymentCreatedEvent(key));
responseWriter.writeEventOnCommand(key, DeploymentIntent.CREATED, event);
}

responseWriter.writeEventOnCommand(key, DeploymentIntent.CREATED, event);
return responseWriter.flush();
});
private void deploymentCreation(long key, long position, DirectBuffer buffer) {
final ActorFuture<Void> pushDeployment =
deploymentDistributor.pushDeployment(key, position, buffer);

actor.runOnCompletion(pushDeployment, (aVoid, throwable) -> writeDeploymentCreatedEvent(key));
}

private void writeDeploymentCreatedEvent(long deploymentKey) {
Expand Down
Loading

0 comments on commit 8cfff92

Please sign in to comment.