Return and accept command topic offsets via REST API #2159

vcrfxia · 2018-11-16T00:10:36Z

Description

This PR adds functionality for users to optionally specify a command topic offset when submitting requests via the REST API. When an offset is provided with a request, the request will not be processed until the server has processed commands from the command topic up through the specified offset. In order to make this functionality meaningful, this PR also augments responses for commands that are successfully written to the command topic to include the offset of the command in the command topic.

Background and motivation: Currently, if a user uses the REST API to interact with a KSQL setup running multiple server threads, there is no guarantee that consecutive requests will hit the same server. This means a user may observe inconsistent responses, for example, if a CREATE STREAM foo (...) request is first received and processed by Server A, and a subsequent INSERT INTO foo (...) request is received by Server B before Server B has fetched and processed the CREATE STREAM command from the command topic. In this scenario, the user would receive an error message reporting that stream foo does not exist.

This PR provides a way to remedy these inconsistencies. When the user sends the CREATE STREAM foo (...) request, the response will include the offset of the CREATE STREAM command in the command topic (assuming the request was successfully parsed and validated). The user can then include this offset as part of their next request, guaranteeing that the server which receives the subsequent request will not try to parse or validate the statement until after it has processed the CREATE STREAM.

Behavior: When request with an offset specified is received, the server will wait no longer than some specified timeout for that offset to be reached before throwing a TimeoutException. This timeout is currently set to be the same as the timeout used when writing commands to the command topic (or the disconnect check interval, in the case of StreamedQueryResource), but this can be updated in a subsequent PR. When no offset is specified with a request (the default), no waiting happens, consistent with the current behavior.

Documentation: Updated documentation accordingly.

Testing done

Added unit tests.

Reviewer checklist

Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
Ensure relevant issues are linked (description should include text like "Fixes #")

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java

ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/CommandStoreTest.java

rodesai

Nice job! Took a pass through and left some comments. There's an accompanying set of changes we need to make to use this functionality from the CLI to give a better experience within a session. But we can get to that in a follow-up.

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStore.java

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/QueuedCommandStatus.java

rodesai · 2018-11-20T19:52:21Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStore.java

@@ -67,6 +75,7 @@ public CommandStore(
    this.commandProducer = commandProducer;
    this.commandIdAssigner = commandIdAssigner;
    this.commandStatusMap = Maps.newConcurrentMap();
+    this.commandOffsetFutures = new ArrayList<>();

    commandConsumer.assign(Collections.singleton(new TopicPartition(commandTopic, 0)));


You can keep this TopicPartition in a member variable and use it when checking the current position. That way we keep the hard-coding of the usage of partition 0 only in 1 place.

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java

rodesai · 2018-11-20T20:01:14Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java

+        throw new RuntimeException(
+            "Interrupted while waiting for command offset of " + String.valueOf(offset), e);
+      } catch (final TimeoutException e) {
+        throw new RuntimeException(


By throwing RuntimeException here, we will return a generic 500 error. We should return a specific error class for this condition, with the appropriate HTTP code.

I'd say the appropriate error code would be a 503 Service Unavailable.

We also need to return a specific error indicating that we timed out waiting for the server to catch up on the command topic. Check out io.confluent.ksql.rest.server.resources.Errors. You can define a new error code like:

public static final int ERROR_CODE_COMMAND_QUEUE_CATCHUP_TIMEOUT = toErrorCode(toErrorCode(SERVICE_UNAVAILABLE.getStatusCode()) + 1); public static Response commandQueueCatchupTimeout(final String msg) { return Response .status(SERVICE_UNAVAILABLE) .entity(new KsqlErrorMessage(ERROR_CODE_COMMAND_QUEUE_CATCHUP_TIMEOUT, msg)) .build(); }

Done. Thanks for the pointers!

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java

ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/KsqlResourceTest.java

rodesai · 2018-11-20T20:40:33Z

ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/KsqlResourceTest.java

+  }
+
+  @Test
+  public void shouldNotWaitIfCommandTopicOffsetReached()


nit: this is a bit of a misleading name, no? The test is actually validating that we do wait until the offset is reached.

This test has changed as result of one of the refactors suggested above, hopefully taking the confusion with it.

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java

big-andy-coates

Thanks @vcrfxia. This is looking good. There are a lot of comments below, but there's actually nothing majorly wrong with what's there already. It's mainly just suggestions and nits.

However, I would suggest two main things:

think more about the status codes being returned by the rest endpoint. 500 INTERNAL_SERVER_ERROR is not appropriate for a timeout, IMHO.
try to avoid 'offset' in the public facing APIs for this. Currently this is implementing using the offset of the Kafka topic. But that's an implementation detail and might change. Better to call it something more generic. Maybe commandId, (though that's currently used else where for something else), or commandNumber or even just correlationId. I'm talking mostly about the fields in KsqlRequest and CommandStatusEntity, but its worth having this flow down all the way to the BlahCommandStore interface and having only the CommandStore impl know its an offset. This way, a different implementation of CommandStore could implement it some other way.

ksql-rest-app/src/main/java/io/confluent/ksql/rest/entity/CommandStatusEntity.java

ksql-rest-app/src/main/java/io/confluent/ksql/rest/entity/KsqlRequest.java

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStore.java

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java

ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/KsqlResourceTest.java

...app/src/test/java/io/confluent/ksql/rest/server/resources/streaming/WSQueryEndpointTest.java

dguy · 2018-11-21T17:24:04Z

I wonder if this could be done by using a similar mechanism to what https://github.com/confluentinc/schema-registry/blob/master/core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaStore.java does? /cc @ewencp

i.e., the command topic writer doesn't write to the topic until the reader has read up to the latest offset. Then we wouldn't need to return an offset that subsequently needs to sent with the next query

ewencp · 2018-11-21T18:15:43Z

I think this makes sense as well. More modern version with new clients is in Connect. Connect has an (internal) utility class that helps manage this (and link goes to particularly relevant part re: end offsets): https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java#L268

rodesai · 2018-11-21T20:03:25Z

@dguy @ewencp what you're describing is a different flavor of the same basic mechanism - it's just that the offset to wait on is inferred to be the current end offset. The thinking behind allowing the client to specify the offset was that a smart client could still partially function even if the cluster is not able to progress on the command topic (e.g. if it can't talk to the schema registry). Any statement that doesn't modify the meta-store could still be run as long as consumption from the command topic is sufficiently far along. Maybe a premature optimization, but it's a pretty small increment over the basic functionality on the server side. Having the client use it correctly is more challenging.

One thing we could do is to accept a designated value (e.g. Long.MAX_LONG) that tells the server to wait until the current end offset. This can just be the default going forward. For now the cli can make a best effort by trying with this value a few times before asking the server not to wait (by omitting the offset) and adding a warning to the output.

vcrfxia

Thanks @rodesai and @big-andy-coates for the very helpful reviews! Made the suggested changes and left a few comments / questions. Notable changes include:

Waiting on futures (and associated error handling) now occurs in CommandStore, rather than having CommandStore return a future that the caller then waits on
QueuedCommandStatus has been split into two classes, QueuedCommandStatus and SequenceNumberFutureStore, to avoid the need for status and command offset setters
Error code on timeout has been updated from 500 INTERNAL_SERVER_ERROR to 503 SERVICE_UNAVAILABLE
Added support for requests with offsets to StreamedQueryResource and WSQueryEndpoint (in addition to KsqlResource)
Renamed “commandOffset” to “commandSequenceNumber”

ksql-rest-app/src/main/java/io/confluent/ksql/rest/entity/CommandStatusEntity.java

ksql-rest-app/src/main/java/io/confluent/ksql/rest/entity/KsqlRequest.java

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStore.java

ksql-rest-app/src/main/java/io/confluent/ksql/rest/entity/CommandStatusEntity.java

ksql-rest-app/src/test/java/io/confluent/ksql/rest/integration/RestApiTest.java

vcrfxia · 2018-11-28T08:06:00Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStatusFuture.java

+  private volatile CommandStatus commandStatus;
+  private final CompletableFuture<CommandStatus> future;
+
+  public CommandStatusFuture(final CommandId commandId) {


This constructor (and the class itself) only need to be public for test purposes (specifically, in KsqlResourceTest), which feels weird to me, but I couldn't find a way around it.

If the test is from another package then you can add a test utility to create instances for you e.g.

package io.confluent.ksql.rest.server.computation public final class CommandStatusFutureTestUtil { ... public static CommandStatusFuture create(final commandId commandId) { return new CommandStatusFuture(commandId); // <- can access as in same package. } }

Good to know. Do you think adding a test util into this package is preferable to leaving the constructor (and class) public? I've updated the other methods in the class to be package-private, so even if someone creates an instance of CommandStatusFuture, they can't do anything with it, which makes me think maybe it's not too bad.

For this class... I don't particularly see an issue with the only constructor being public.

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandRunner.java

...est-app/src/test/java/io/confluent/ksql/rest/server/resources/StreamedQueryResourceTest.java

docs/developer-guide/api.rst

JimGalasyn

LGTM, with one suggested tweak.

Co-Authored-By: vcrfxia <[email protected]>

rodesai · 2018-12-01T01:05:23Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java

@@ -190,6 +191,9 @@ public Response handleKsqlStatements(final KsqlRequest request) {
    activenessRegistrar.updateLastRequestTime();

    try {
+      CommandStoreUtil.httpWaitForCommandSequenceNumber(


why the prefix http?

CommandStoreUtil has two versions of waitForCommandSequenceNumber, waitForCommandSequenceNumber which potentially throws a TimeoutException (and is called by WSQueryEndpoint), and httpWaitForCommandSequenceNumber which catches the TimeoutException and wraps it into a KsqlRestException (and is called by KsqlResource and StreamedQueryResource). If this function name is confusing, we can easily rename it. Got any suggestions?

I wouldn't have the two versions personally. But then again, I don't think this warrants a utility class. I would just inline in WSQueryEndpoint like:

try { request.getCommandSequenceNumber() .ifPresent(seqNo -> replayableCommandQueue.ensureConsumedUpThrough(seqNo, timeout)); } catch (final TimeoutException e) { log.debug("Timed out wait for prior commands to execute. seqNo: " + seqNo, e); SessionUtil.closeSilently(session, CloseCodes.TRY_AGAIN_LATER, e.getMessage()); }

And in KsqlResource and StreamedQueryResource I would move the logic into a function within those classes. There just isn't enough code here to warrant a utility class. The fact you're changing exception types indicates to me it should be in the calling class.

private void waitForPriorCommandToExecute(final KsqlRequest request) { try { request.getCommandSequenceNumber().ifPresent(seqNo -> replayableCommandQueue.ensureConsumedUpThrough(seqNo, disconnectCheckInterval)); } catch (final TimeoutException e) { throw new KsqlRestException(Errors.commandQueueCatchUpTimeout(e.getMessage())); } }

Leaving this utility for now. (See explanation below.)

...est-app/src/main/java/io/confluent/ksql/rest/server/resources/streaming/WSQueryEndpoint.java

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandRunner.java

rodesai · 2018-12-03T08:01:25Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/client/KsqlRestClient.java

@@ -132,7 +132,7 @@ public void setServerAddress(final String serverAddress) {
  }

  public RestResponse<KsqlEntityList> makeKsqlRequest(final String ksql) {


We need to extend the APIs here to use the offset mechanism. Fine to add that in a follow-up though.

Yup, will do in a follow-up.

ksql-rest-app/src/main/java/io/confluent/ksql/rest/entity/CommandStatusEntity.java

...t-app/src/main/java/io/confluent/ksql/rest/server/computation/SequenceNumberFutureStore.java

...p/src/main/java/io/confluent/ksql/rest/server/resources/streaming/StreamedQueryResource.java

rodesai · 2018-12-03T08:36:45Z

ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/CommandStoreTest.java

+    // Given:
+    givenCmdStoreUpThroughPosition(2);
+    expect(sequenceNumberFutureStore.getFutureForSequenceNumber(EasyMock.anyLong())).andReturn(future);
+    expect(future.get(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class))).andReturn(null);


We should verify that the right timeout is used.

Done. (Also cleaned up and refactored the test to Mockito while I was it.)

...est-app/src/test/java/io/confluent/ksql/rest/server/resources/StreamedQueryResourceTest.java

rodesai · 2018-12-03T09:00:27Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStore.java

+  public void ensureConsumedUpThrough(final long seqNum, final long timeout)
+      throws TimeoutException {
+    final long consumerPosition = getNextConsumerSequenceNumber();
+    if (consumerPosition > seqNum) {


I think this is not quite what we want here. consumerPosition is the consumer's position, which is the offset of the next record to be consumed. What we want here is the offset of the last command that we know we processed. These are not the same for the current batch of records being processed. Imagine the executor is off processing a batch of records consisting of a single record w/ offset 10. consumerPosition would be 11, so we might not wait for seqNum=10, when we should, since we don't know if its been completed yet. So what we really want is our own bookkeeping of the latest processed command.

I also think you should move this bookkeeping and check into SequenceNumberFutureStore and add some synchronization with completeFuturesUpToSequenceNumber. The reason is that there's a small race here where an API thread could call ensureConsumedUpThrough and the statement executor thread could be just finishing the latest batch of statements. If it completes futures and sleeps on consume between the check and registering the future, this could timeout when it shouldn't.

Good catch! Done.

vcrfxia

Thanks for the review @rodesai ! Made the requested changes.

ksql-rest-app/src/main/java/io/confluent/ksql/rest/entity/CommandStatusEntity.java

vcrfxia · 2018-12-03T19:40:06Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/client/KsqlRestClient.java

@@ -132,7 +132,7 @@ public void setServerAddress(final String serverAddress) {
  }

  public RestResponse<KsqlEntityList> makeKsqlRequest(final String ksql) {


Yup, will do in a follow-up.

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandRunner.java

...p/src/main/java/io/confluent/ksql/rest/server/resources/streaming/StreamedQueryResource.java

...est-app/src/main/java/io/confluent/ksql/rest/server/resources/streaming/WSQueryEndpoint.java

...est-app/src/test/java/io/confluent/ksql/rest/server/resources/StreamedQueryResourceTest.java

vcrfxia · 2018-12-03T21:32:58Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStore.java

+  public void ensureConsumedUpThrough(final long seqNum, final long timeout)
+      throws TimeoutException {
+    final long consumerPosition = getNextConsumerSequenceNumber();
+    if (consumerPosition > seqNum) {


Good catch! Done.

...t-app/src/main/java/io/confluent/ksql/rest/server/computation/SequenceNumberFutureStore.java

vcrfxia · 2018-12-03T21:41:30Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java

@@ -190,6 +191,9 @@ public Response handleKsqlStatements(final KsqlRequest request) {
    activenessRegistrar.updateLastRequestTime();

    try {
+      CommandStoreUtil.httpWaitForCommandSequenceNumber(


CommandStoreUtil has two versions of waitForCommandSequenceNumber, waitForCommandSequenceNumber which potentially throws a TimeoutException (and is called by WSQueryEndpoint), and httpWaitForCommandSequenceNumber which catches the TimeoutException and wraps it into a KsqlRestException (and is called by KsqlResource and StreamedQueryResource). If this function name is confusing, we can easily rename it. Got any suggestions?

big-andy-coates

Thanks @vcrfxia and sorry for the review delay. Looking real close. Just a few things... ;)

docs/developer-guide/api.rst

ksql-rest-app/src/main/java/io/confluent/ksql/rest/entity/CommandStatusEntity.java

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStatusFuture.java

big-andy-coates · 2018-12-04T16:43:32Z

ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/KsqlResourceTest.java

+  }
+
+  @Test
+  public void shouldReturn503IfTimeoutWhileWaitingForCommandSequenceNumber() throws Exception {


Suggested change

public void shouldReturn503IfTimeoutWhileWaitingForCommandSequenceNumber() throws Exception {

public void shouldReturnServiceUnavailableIfTimeoutWaitingForCommandSequenceNumber() throws Exception {

ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/KsqlResourceTest.java

...est-app/src/test/java/io/confluent/ksql/rest/server/resources/StreamedQueryResourceTest.java

rodesai · 2018-12-04T21:39:49Z

...est-app/src/main/java/io/confluent/ksql/rest/server/resources/streaming/WSQueryEndpoint.java

+            commandQueueCatchupTimeout);
+      } catch (final TimeoutException e) {
+        log.debug("Timeout while processing request", e);
+        SessionUtil.closeSilently(session, CloseCodes.TRY_AGAIN_LATER, e.getMessage());


Missing a return here. There should be a test that's failing.

Yikes, good catch. Added a check for this into WSQueryEndpointTest#shouldReturnErrorIfCommandQueueCatchupTimeout().

rodesai

LGTM, once the in-line comment about the missing return is addressed. Thanks!

...t-app/src/main/java/io/confluent/ksql/rest/server/computation/SequenceNumberFutureStore.java

vcrfxia

Thanks for the second round of reviews, @big-andy-coates and @rodesai ! Gave it another pass.

ksql-rest-app/src/main/java/io/confluent/ksql/rest/entity/CommandStatusEntity.java

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStore.java

...p/src/main/java/io/confluent/ksql/rest/server/resources/streaming/StreamedQueryResource.java

vcrfxia · 2018-12-04T20:41:52Z

ksql-rest-app/src/test/java/io/confluent/ksql/rest/entity/CommandStatusEntityTest.java

+import io.confluent.ksql.rest.server.computation.CommandId;
+import org.junit.Test;
+
+public class CommandStatusEntityTest {


Added tests checking null values in the constructor. Don't think additional serialization/deserialization tests are warranted, but let me know if you disagree.

ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/CommandStoreTest.java

...rest-app/src/main/java/io/confluent/ksql/rest/server/computation/ReplayableCommandQueue.java

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStore.java

vcrfxia · 2018-12-05T23:13:14Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStatusFuture.java

+  private volatile CommandStatus commandStatus;
+  private final CompletableFuture<CommandStatus> future;
+
+  public CommandStatusFuture(final CommandId commandId) {


Good to know. Do you think adding a test util into this package is preferable to leaving the constructor (and class) public? I've updated the other methods in the class to be package-private, so even if someone creates an instance of CommandStatusFuture, they can't do anything with it, which makes me think maybe it's not too bad.

docs/developer-guide/api.rst

big-andy-coates · 2018-12-13T12:11:56Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStatusFuture.java

+    try {
+      return finalStatusFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
+    } catch (final ExecutionException e) {
+      throw new RuntimeException("Error executing command " + commandId, e.getCause());


nit: don't wrap in RuntimeException if already one

Suggested change

throw new RuntimeException("Error executing command " + commandId, e.getCause());

if (e instanceof RuntimeException) {

throw (RuntimeException)e;

}

throw new RuntimeException("Error executing command " + commandId, e.getCause());

big-andy-coates · 2018-12-13T12:21:24Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStatusFuture.java

+  private volatile CommandStatus commandStatus;
+  private final CompletableFuture<CommandStatus> future;
+
+  public CommandStatusFuture(final CommandId commandId) {


For this class... I don't particularly see an issue with the only constructor being public.

big-andy-coates · 2018-12-13T12:28:53Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStore.java

@@ -44,31 +50,53 @@
 * the beginning until now, or any new messages since then), and writing to it.
 */

+// CHECKSTYLE_RULES.OFF: ClassDataAbstractionCoupling


Do you still need to turn this rule off with Hojjat's work to split the CommandTopic out?

Shouldn't be, but I'm merging first. Will coordinate with @hjafarpour to have to it removed.

ksql-rest-app/src/main/java/io/confluent/ksql/rest/util/CommandStoreUtil.java

...t-app/src/main/java/io/confluent/ksql/rest/server/computation/SequenceNumberFutureStore.java

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java

big-andy-coates

.

big-andy-coates

Thanks @vcrfxia - LGTM, except ...

Thinking more on @dguy 's suggestion: #2159 (comment) and then @ewencp's and @rodesai's follow on comments...

The default behaviour, if no command sequence is provided in the request, is to do nothing special right? i.e. don't wait for any previous request to finish. This means the default behaviour can result in requests failing if a previous request, upon which it relies, hasn't been fully handled.

I think a safer default is to wait for all previous requests to have finished. Then more advanced users can use the sequence number to improve throughput for unrelated requests.

It probably makes sense to do this in a follow up PR... so created #2268 and assigned to you.

JimGalasyn

LGTM!

vcrfxia

Sure @big-andy-coates , I'll look at incorporating that into my follow-up PR that will introduce that as the default behavior for the CLI.

Thanks for the review! Added your requested tests/checks. Will merge once the build passes.

...t-app/src/main/java/io/confluent/ksql/rest/server/computation/SequenceNumberFutureStore.java

vcrfxia · 2018-12-13T19:03:01Z

ksql-rest-app/src/test/java/io/confluent/ksql/rest/entity/CommandStatusEntityTest.java

+import io.confluent.ksql.rest.server.computation.CommandId;
+import org.junit.Test;
+
+public class CommandStatusEntityTest {


vcrfxia · 2018-12-13T19:04:30Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStore.java

@@ -44,31 +50,53 @@
 * the beginning until now, or any new messages since then), and writing to it.
 */

+// CHECKSTYLE_RULES.OFF: ClassDataAbstractionCoupling


Shouldn't be, but I'm merging first. Will coordinate with @hjafarpour to have to it removed.

* fork/master: (107 commits) copy-edited (confluentinc#2299) copy-edited (confluentinc#2299) remove unused map from command store `getRestoreCommands` (confluentinc#2296) Speed up integration tests. (confluentinc#2288) Move KSQL Clickstream demo to Examples repo (confluentinc#2270) Bump Confluent to 5.1.1-SNAPSHOT, Kafka to 2.1.1-SNAPSHOT Set Confluent to 5.1.0, Kafka to 2.1.0-cp1. Add ServiceContext (confluentinc#2243) Update to use CCL (confluentinc#2278) Remove redundant intro sentence in README.md (confluentinc#2277) Switch to use CCL (confluentinc#2275) Update readme for CCL (confluentinc#2276) Minor: add Hamcrest matchers for KsqlRestException and KsqlErrorMessage (confluentinc#2273) Update per relicensing (confluentinc#2274) CP-584: 5.1.0 changelog (confluentinc#2267) add null checks to min, max, and count aggregates; call Math.min/max for consistency (confluentinc#2246) Return and accept command topic offsets via REST API (confluentinc#2159) MINOR: Fix bug encountered when restoring RUN SCRIPT command. (again) (confluentinc#2265) More improvements to the way the CLI handles inline comments in multi-line statements. (confluentinc#2241) Fix issue where Avro subject does not exist (confluentinc#2260) ...

vcrfxia added 2 commits November 15, 2018 14:34

return command topic offset with success response from KsqlResource.

c3a70c9

accept offset to wait for in KsqlRequest.

7241858

vcrfxia requested a review from a team as a code owner November 16, 2018 00:10

vcrfxia commented Nov 16, 2018

View reviewed changes

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java Outdated Show resolved Hide resolved

vcrfxia commented Nov 16, 2018

View reviewed changes

ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/CommandStoreTest.java Outdated Show resolved Hide resolved

vcrfxia changed the title ~~Request indices~~ Return and accept command topic offsets via REST API Nov 16, 2018

rodesai reviewed Nov 20, 2018

View reviewed changes

big-andy-coates reviewed Nov 20, 2018

View reviewed changes

vcrfxia added 2 commits November 27, 2018 23:14

address comments.

baffaf8

rename 'offset' to 'sequence number'.

8890ad9

vcrfxia commented Nov 28, 2018

View reviewed changes

vcrfxia requested review from rodesai and big-andy-coates November 28, 2018 08:28

vcrfxia added 4 commits November 28, 2018 09:50

Merge master.

838e102

Merge branch 'master' into request-indices

4606a96

Merge branch 'master' into request-indices

fcc4a82

add StreamQueryResource tests for new behavior.

b343130

apurvam mentioned this pull request Nov 30, 2018

UX: CSAS returns control to UI before completing, subsequent statement fails with Unable to verify the AVRO schema is compatible with KSQL. Subject not found. #1394

Closed

add documentation.

e9af7fc

vcrfxia requested a review from JimGalasyn as a code owner November 30, 2018 18:31

JimGalasyn reviewed Nov 30, 2018

View reviewed changes

docs/developer-guide/api.rst Outdated Show resolved Hide resolved

JimGalasyn approved these changes Nov 30, 2018

View reviewed changes

jim's requested edit.

d89059e

Co-Authored-By: vcrfxia <[email protected]>

vcrfxia requested a review from a team November 30, 2018 20:49

rodesai reviewed Dec 3, 2018

View reviewed changes

rohan's requested changes.

44f5b28

vcrfxia commented Dec 3, 2018

View reviewed changes

refactor CommandStoreTest to Mockito.

3463a22

big-andy-coates reviewed Dec 4, 2018

View reviewed changes

rodesai reviewed Dec 4, 2018

View reviewed changes

rodesai approved these changes Dec 5, 2018

View reviewed changes

...t-app/src/main/java/io/confluent/ksql/rest/server/computation/SequenceNumberFutureStore.java Show resolved Hide resolved

latest requested changes.

c516723

vcrfxia commented Dec 5, 2018

View reviewed changes

minor requested change that slipped through the cracks.

8e15a75

vcrfxia requested review from big-andy-coates and JimGalasyn December 5, 2018 23:45

Merge branch 'master' into request-indices

64fca27

big-andy-coates reviewed Dec 13, 2018

View reviewed changes

big-andy-coates approved these changes Dec 13, 2018

View reviewed changes

JimGalasyn approved these changes Dec 13, 2018

View reviewed changes

vcrfxia added 3 commits December 13, 2018 11:10

final round of andy's requested changes.

c0f9886

Merge branch 'master' into request-indices

aa2d38a

fix checkstyle issue.

807cbd5

vcrfxia commented Dec 13, 2018

View reviewed changes

vcrfxia merged commit 20d6512 into confluentinc:master Dec 13, 2018

vcrfxia deleted the request-indices branch December 13, 2018 21:03

vcrfxia added this to the 5.2 milestone Feb 19, 2019

		@@ -132,7 +132,7 @@ public void setServerAddress(final String serverAddress) {
		}

		public RestResponse<KsqlEntityList> makeKsqlRequest(final String ksql) {

	public void shouldReturn503IfTimeoutWhileWaitingForCommandSequenceNumber() throws Exception {
	public void shouldReturnServiceUnavailableIfTimeoutWaitingForCommandSequenceNumber() throws Exception {

Return and accept command topic offsets via REST API #2159

Return and accept command topic offsets via REST API #2159

Conversation

vcrfxia commented Nov 16, 2018 • edited Loading

Description

Testing done

Reviewer checklist

rodesai left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

big-andy-coates left a comment • edited Loading

Choose a reason for hiding this comment

dguy commented Nov 21, 2018

ewencp commented Nov 21, 2018

rodesai commented Nov 21, 2018 • edited Loading

vcrfxia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

JimGalasyn left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vcrfxia Dec 5, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vcrfxia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

big-andy-coates left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rodesai Dec 4, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rodesai left a comment • edited Loading

Choose a reason for hiding this comment

vcrfxia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

big-andy-coates left a comment • edited Loading

Choose a reason for hiding this comment

big-andy-coates left a comment

Choose a reason for hiding this comment

JimGalasyn left a comment

Choose a reason for hiding this comment

vcrfxia left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vcrfxia commented Nov 16, 2018 •

edited

Loading

big-andy-coates left a comment •

edited

Loading

rodesai commented Nov 21, 2018 •

edited

Loading

vcrfxia Dec 5, 2018 •

edited

Loading

rodesai Dec 4, 2018 •

edited

Loading

rodesai left a comment •

edited

Loading

big-andy-coates left a comment •

edited

Loading

vcrfxia left a comment •

edited

Loading