feat: Transactional Produces to Command Topic #3660

stevenpyzhang · 2019-10-24T00:37:48Z

Description

Protocol

If a statement needs to be distributed (in DistributingExecutor):
Initialize a new producer with a fixed value (based on ksql.service.id) for transactions.id. All KSQL request handlers must always use the same value for transactions.id. This ensures that Request Handlers that already started a transaction are unable to commit once a given producer's transaction is in-flight.
Start a transaction 
Wait for the command topic consumer to consume up to the last offset (we need to ask kafka for the last offset after starting the transaction.)
Validate the request
Produce the command to the command topic
Commit the transaction

We want to ensure that every command written to the commandTopic is being validated only after all previous commands have been processed in order to have the most up to date snapshot of the KsqlEngine for validation purposes.

Some notes

I removed the existing CommandProducer from CommandTopic.java
An entire transaction is in DistributingExecutor
A new consumer is created every time we waitForCommandConsumer in order to get the end offset of the command topic, this can't be done with the existing Consumer in CommandTopic.java because it's running in the CommandRunner thread, which is separate from the thread handling requests in KsqlResource (consumers are not thread-safe)
-new configs added to improve durability of command topic, set directly in code unless otherwise specified.

//for durability of command topic
ksql.internal.topic.replication.factor=3 //passed in through configs
ksql.internal.topic.min.insync.replicas=2 //passed in through configs
unclean.leader.election.enable=false

producer.acks=all // only for the command topic producer

//to enable transactions
isolation.level="read_commited" //consumer
transactional.id=[ksql_server_id] //producer

These changes also require additional ACLs to be set including

WRITE TRANSACTIONAL_ID "<ksql.service.id>": A transactional producer which has its transactional.id set requires this privilege.

DESCRIBE TRANSACTIONAL_ID "<ksql.service.id>": This applies only on transactional producers and checked when a producer tries to find the transaction coordinator.

DESCRIBE TOPIC "__consumer_offsets" "__transaction_state"

Follow up to this PR:
#3795
#3768

Transactional Producer API Docs
https://kafka.apache.org/10/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html

Kip introducing transactions
https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging#KIP-98-ExactlyOnceDeliveryandTransactionalMessaging-TestPlan

More on Transactional Messaging in Kafka
https://cwiki.apache.org/confluence/display/KAFKA/Transactional+Messaging+in+Kafka

Testing done

Existing tests fixed, may need a new integration test to ensure functionality.

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/execution/RequestHandler.java

stevenpyzhang · 2019-10-24T00:40:59Z

ksql-test-util/src/main/java/io/confluent/ksql/test/util/EmbeddedSingleNodeKafkaCluster.java

@@ -470,6 +470,10 @@ private Properties buildBrokerConfig(final String logDir) {
    config.put(KafkaConfig.LogRetentionTimeMillisProp(), -1);
    // Stop logs marked for deletion from being deleted
    config.put(KafkaConfig.LogDeleteDelayMsProp(), Long.MAX_VALUE);
+    // Set to 1 because only 1 broker


Without setting these, this error appears for some integration tests

ERROR [KafkaApi-0] Number of alive brokers '1' does not meet the required replication factor '3' for the transactions state topic (configured via 'transaction.state.log.replication.factor'). This error can be ignored if the cluster is starting up and not all brokers are up yet. (kafka.server.KafkaApis:74)```

purplefox

Hey Steven, thanks for this!
I'm going to start a review, but I wanted to make sure I understand the motivations for this PR before I start:

We want to ensure that every statement written to the commandTopic is being validated after every statement before it has been processed before in order to have the most up to date snapshot of the KsqlEngine for validation purposes.

Could you elaborate a bit on the above statement? I'm trying to understand why this is important. Perhaps you could illustrate with an example?

stevenpyzhang · 2019-10-24T21:08:51Z

Could you elaborate a bit on the above statement? I'm trying to understand why this is important. Perhaps you could illustrate with an example?

@purplefox , Currently, before we produce commands to the command topic, we do validation on the statement to ensure it can be executed properly. And when we're actually executing the statement on a KSQL server after consuming it from the command topic, we're also doing these validations.

There's two main forms of validations being done, internal state and external state. For example, a Create Stream FooStream... statement, one Internal state validation would be ensuring no stream already exists with the name FooStream while an external state validation might be checking to see if the underlying Kafka Topic Foo exists.

We're trying to move away from doing validation when executing the statement from the command topic.
The work being done with Klip-6 https://github.com/confluentinc/ksql/blob/master/design-proposals/klip-6-execution-plans.md eliminates the need for external validations. This PR would eliminated the need for doing internal state validations when executing.

A simple example of something that this code would affect is currently if two KSQL servers concurrently try to produce the same CREATE STREAM FOO to the command topic, both would be enqueued. One of the statements would fail when it's actually executed. Although in this example the final state of the KSQL server is fine (a Stream named FOO is created), It would be better if we can ensure every statement written to the command topic at offset n has been validated against a KSQL internal state that has processed all statements up to n-1 (the internal KSQL state is most up to date).

There's some discussion in this issue that would also be helpful in getting context. Also includes some more examples
#2435

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

purplefox · 2019-10-24T21:34:48Z

Thanks Steven, I think I have a clearer picture of the motivations here now.

It would be better if we can ensure every statement written to the command topic at offset n has been validated against a KSQL internal state that has processed all statements up to n-1 (the internal KSQL state is most up to date).

This seems to imply that every time we add anything to the command topic we must first wait for all outstanding commands to be processed locally.

This means we're going to incur, at least a round trip time to the Kafka broker for each command, i.e. our throughput we will be limited by the latency of ksql<->kafka. If that's measured in milliseconds it means our theoretical maximum throughput will be in the 100s of transactions of second, which seems very low.

I'm wondering if we really want to impose such a bottleneck on the system?

purplefox

Looks good! Just a few comments.

purplefox · 2019-10-24T23:24:23Z

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

@@ -646,7 +668,8 @@ private static void maybeCreateProcessingLogStream(
      return;
    }

-    commandQueue.enqueueCommand(configured.get());
+    commandQueue.enqueueCommand(configured.get(), producerTransactionManager);
+    producerTransactionManager.commit();


Should we also be catching exceptions and aborting the transaction on failure?

Transactions have been isolated to DistributingExecutor and will be aborted if an exception is thrown.

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

purplefox · 2019-10-24T23:28:18Z

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

+
+    try {
+      int retries = 0;
+      while (commandRunner.getNumCommandProcessed() < endOffset) {


Instead of polling getNumCommandProcessed() we could add a method to commandRunner waitUntilOffset(long offset) which didn't return until the required offset was reached. Internally it could use Object.wait() and Object.notify() to wait for the required condition.

Is this assuming that every offset in the command topic corresponds to a command? I thought that with transactions these need not necessarily be in sync since transactions write additional messages into the topic (which take up offsets but aren't commands), though I'm having a surprisingly tough time finding docs about this. Would be good to have someone else confirm or deny.

When I was testing this locally, I was able to make several transactions in a row without this method getting stuck. If additional messages were taking up offsets, I'd hit the retry limit and have an exception thrown which didn't happen. So that behavior makes me think it doesn't effect commandConsumer.endOffsets() but someone more familiar with this should chime in. I can't seem to find any documentation about these additional messages either.

Every transaction will write at least one control message to each partition in the transaction. This control message (commit or abort message) will have an offset. I'm actually surprised that your local testing worked.

Anothre risk is if we have aborted transactions: and there is a chance of these if an active produce gets fenced by a newer instance coming on line. In this case, the number of valid (non-aborted commands) will always be divergent from the end offsets.

The docs on how this works can be found here https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging

I ended up reusing the functionality we already have in place for commandStore.ensureConsumedPast() which uses offsets. position() will get the next offset to fetch so this will ignore the control messages

purplefox · 2019-10-24T23:31:31Z

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

+      }
+      throw new RuntimeException(e.getCause());
+    } catch (final InterruptedException e) {
+      throw new RuntimeException(e);


Should we not abort the tx on all exceptions?

the exception should bubble up and get handled in DistributingExecutor where the transaction would be aborted

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

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/execution/RequestHandler.java

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

rodesai · 2019-10-27T19:09:53Z

I'm wondering if we really want to impose such a bottleneck on the system?

The reason I'm not concerned about this is that we only need to follow the protocol implemented here for statements that need to be logged to the command topic (CREATE STREAM, CREATE TABLE, CREATE STREAM AS, CREATE TABLE AS, DROP, TERMINATE). We don't expect a significant rate of these on any KSQL cluster.

vinothchandar

Further, topics which are included in transactions should be configured for durability. In particular, the replication.factor should be at least 3, and the min.insync.replicas for these topics should be set to 2. Finally, in order for transactional guarantees to be realized from end-to-end, the consumers must be configured to read only committed messages as well.

As a part of this effort, could we also introduce configs for the ksql internal topics (i.e command topic) be configured for durability. When I was looking at something else, noticed that we only support setting "ksql.internal.topic.replicas" , which may not be sufficient..

vinothchandar · 2019-10-28T18:49:26Z

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

@@ -646,7 +668,8 @@ private static void maybeCreateProcessingLogStream(
      return;
    }

-    commandQueue.enqueueCommand(configured.get());
+    commandQueue.enqueueCommand(configured.get(), transactionalProducer);
+    transactionalProducer.commit();


is the exception handling and the abortTransaction() handled inside enqueueCommand > Transactional Producer::send? could we avoid calling commit() for an aborted transaction ?

rodesai · 2019-10-28T21:06:27Z

I'm wondering if we really want to impose such a bottleneck on the system?

The reason I'm not concerned about this is that we only need to follow the protocol implemented here for statements that need to be logged to the command topic (CREATE STREAM, CREATE TABLE, CREATE STREAM AS, CREATE TABLE AS, DROP, TERMINATE). We don't expect a significant rate of these on any KSQL cluster.

Ah I see the current implementation does this wait for all commands. We should fix that.

stevenpyzhang · 2019-10-28T21:15:49Z

I'm wondering if we really want to impose such a bottleneck on the system?

The reason I'm not concerned about this is that we only need to follow the protocol implemented here for statements that need to be logged to the command topic (CREATE STREAM, CREATE TABLE, CREATE STREAM AS, CREATE TABLE AS, DROP, TERMINATE). We don't expect a significant rate of these on any KSQL cluster.

Ah I see the current implementation does this wait for all commands. We should fix that.

Should the TransactionalProducer be moved to DistributingExecutor then? It would help get rid of that injection pattern I currently have and it makes sense since DistributingExecutor would be the only one that uses it.

I think the main problem though is that the transaction needs to be started before the RequestValidator validation happens. If there's CREATE STREAM, CREATE TABLE, etc...
mixed in with non-distributed statements like SHOW TOPICS, I think it only makes sense to wait beforehand for the commandRunner and then validate all the statements.

rodesai · 2019-10-28T21:58:32Z

Should the TransactionalProducer be moved to DistributingExecutor then? It would help get rid of that injection pattern I currently have and it makes sense since DistributingExecutor would be the only one that uses it.

I think the main problem though is that the transaction needs to be started before the RequestValidator validation happens. If there's CREATE STREAM, CREATE TABLE, etc...
mixed in with non-distributed statements like SHOW TOPICS, I think it only makes sense to wait beforehand for the commandRunner and then validate all the statements.

I think the current structure is reasonable (other than the way the producer is injected - but I'll leave that for the code comments) - the distributing executor doesn't need to know about validation. It would definitely be worthwhile to optimize away creating the producer/consumer, and checking offsets for statements that don't need to distribute to the command topic though.

stevenpyzhang · 2019-10-28T22:04:13Z

other than the way the producer is injected

One option I thought of for getting rid of injecting the TransactionalProducer into DistributingExecutor could be to have it implement a DistributedStatementExecutor interface which extends StatementExectuor. DistributedStatementExecutor interface would have an execute function with the required arg types for passing in the producer.

stevenpyzhang · 2019-10-28T22:07:51Z

optimize away creating the producer/consumer

@rodesai I thought during a discussion we had about the implementation details of this, you mentioned that we needed to create a new Producer for each transaction since there could be multiple threads on a server trying to produce to the command topic. I believe the consumers for getting offset could be optimized to just one consumer that's shared among all the TransactionalProducers (passed to each new TransactionalProducer from the factory class).

rodesai · 2019-10-28T22:20:37Z

optimize away creating the producer/consumer

@rodesai I thought during a discussion we had about the implementation details of this, you mentioned that we needed to create a new Producer for each transaction since there could be multiple threads on a server trying to produce to the command topic. I believe the consumers for getting offset could be optimized to just one consumer that's shared among all the TransactionalProducers (passed to each new TransactionalProducer from the factory class).

I just mean optimizing it away when handling requests that don't need a transaction (e.g. list streams;, or a pull query). For requests that include statements that need to be distributed let's just optimize for keeping the code as simple as possible, and not worry about creating a new consumer/producer each time for these.

rodesai · 2019-10-29T13:13:35Z

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

+import java.util.Objects;
+import org.apache.kafka.clients.producer.ProducerConfig;
+
+public class TransactionalProducerFactory {


Rather than having a separate class for this, I'd just add an interface to CommandQueue like createTransactionalProducer, and drop enqueueCommand from the CommandQueue interface. Also, to avoid breaking the current abstraction (KsqlResource/RequestHandler just talk to a CommandQueue) we should have an interface for TransactionalProducer:

interface TransactionalProducer { void begin(); void waitForConsumer(); void QueuedCommandStatus send(final CommandId commandId, final Command command); // here send would just return the offset void abort(); void commit(); void close(); }

The current TransactionProducer can move to an implementation of this interface and be created by CommandStore when createTransactionalProducer is called.

commandStore would need to keep track of the commandRunner in order to pass it to each new TransactionalProducer in createTransactionalProducer. I'm not sure how exactly that would work out because the commandStore is created first in RestApp and then the commandRunner.

Why can't we reuse the existing mechanism in CommandStore to do this? (see the implementation of completeSatisfiedSequenceNumberFutures). Then the producer factory has no dependency on CommandRunner.

The CommandStore version introduces a race condition. The futures for the sequenceNumbers are completed before the commands are actually executed in commandRunner. There's a bit of lag between consuming and actually executing the command and updating the metastore. Also, the CommandRunner could get stuck executing consumed commands, but ensureConsumedPast() in CommandStore wouldn't block the TransactionalProducer.waitForConsumer() call and then the transaction would proceed on an out of sync metastore.

After thinking about it and playing around with the existing functionality of CommandStore.ensureComsumedPast(), it fits with what I was trying to achieve by duplicating the code. I removed the code from CommandRunner and the factory class. I'm not sold on dropping enqueueCommand() from commandQueue though since the CommandStore.commandStatusMap is being updated in enqueueCommand().

I'm not sold on dropping enqueueCommand() from commandQueue though since the CommandStore.commandStatusMap is being updated in enqueueCommand()

Is the concern that you won't be able to get to the map from the queue? In this way of doing things the implementation of the producer interface could be inside CommandStore, so that shouldn't be a problem. So something like:

interface TransactionalProducer { send() } interface CommandQueue { getProducer() } class CommandStore { Producer getProducer() { return new TopicProducer(); } class TopicProducer implements TransactionalProducer { send() { ... } } }

This way, the underlying queue is still encapsulated away.

Mind if I leave this for a future PR? I think this PR has gone through enough refactors already.

rodesai · 2019-10-29T13:14:50Z

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

+
+      final QueuedCommandStatus queuedCommandStatus =
+          commandQueue.enqueueCommand(injected, transactionalProducer);
+
      final CommandStatus commandStatus = queuedCommandStatus


We can't actually wait for the status here any more, since the consumer thread will not have read the command yet. Instead, we should wait for the status at the end once all the commands have been enqueued. But, that raises another issue - we might have statements that read the meta-store (eg list streams or describe) after a logged command. These statements should see the changes applied by the logged command. So we either need to:
- maintain a shadow engine clone and apply the commands to it as we go along, and then execute statements that don't need to be logged against the clone.
- run the validate-execute-commit loop for each statement one-by-one (rather than validating them all up front).

My vote is for the latter option, but we should see what other folks think (cc @big-andy-coates @agavra)

We could also re-run validation for logged statements as part of execute. So DistributingExecutor would just run the whole protocol (init txn, validate, log cmd, commit). It's a bit wasteful to re-run validation, but I think this is the simplest way to get this PR unblocked.

maintain a shadow engine clone and apply the commands to it as we go along, and then execute statements that don't need to be logged against the clone.

This option could potentially return misleading results in the event of an abortTransaction();

//starting from no streams CREATE STREAM foo1(age BIGINT) WITH (KAFKA_TOPIC='foo', VALUE_FORMAT='JSON'); CREATE STREAM foo2(age BIGINT) WITH (KAFKA_TOPIC='foo', VALUE_FORMAT='JSON'); show streams; CREATE STREAM foo3(age BIGINT) WITH (KAFKA_TOPIC='foo', VALUE_FORMAT='JSON');

If the first two messages are sent and applied to the shadow engine, list streams executed against the shadow engine would return that stream foo1 and foo2 have been created. If sending foo3 fails, the transaction wouldn't be committed and the commandRunner never processes any of the 3 Create Stream statements. Running list streams again would show that the server never actually created the streams.

run the validate-execute-commit loop for each statement one-by-one (rather than validating them all up front).

I think there's also issues with this. With how the code is currently designed, a KsqlEntityList is only returned if all the statements are successfully executed.

ksql> CREATE TABLE foo400(age BIGINT) WITH (KAFKA_TOPIC='foo', VALUE_FORMAT='JSON'); >CREATE TABLE foo200(age BIGINT) WITH (KAFKA_TOPIC='foo', VALUE_FORMAT='JSON'); >Show Tables; CREATE TABLE foo400(age BIGINT) WITH (KAFKA_TOPIC='foo', VALUE_FORMAT='JSON'); Message --------------- Table created --------------- CREATE TABLE foo200(age BIGINT) WITH (KAFKA_TOPIC='foo', VALUE_FORMAT='JSON'); Message --------------- Table created --------------- Show Tables; Table Name | Kafka Topic | Format | Windowed ---------------------------------------------- FOO200 | foo | JSON | false FOO400 | foo | JSON | false ---------------------------------------------- ksql> ksql> CREATE TABLE foo100(age BIGINT) WITH (KAFKA_TOPIC='foo', VALUE_FORMAT='JSON'); >CREATE TABLE foo200(age BIGINT) WITH (KAFKA_TOPIC='foo', VALUE_FORMAT='JSON'); >Show Tables; Cannot add table 'FOO200': A table with the same name already exists ksql> show tables; Table Name | Kafka Topic | Format | Windowed ---------------------------------------------- FOO100 | foo | JSON | false FOO200 | foo | JSON | false FOO400 | foo | JSON | false ---------------------------------------------- ksql>

The code would need to be refactored to be able to handle processing individual commands (which should probably be done before implementing the TransactionalProduces)

This option could potentially return misleading results in the event of an abortTransaction();

In this case we'd just fail the whole request - so no response to list streams;

The code would need to be refactored to be able to handle processing individual commands (which should probably be done before implementing the TransactionalProduces)

I'm not sure I follow. It's possible today for any command to fail when executing, in which case we just fail the whole command. We could do the same thing if validation inside the transaction fails. I agree it's better to return the partial response with the results for what we could successfully execute. My point is that this is orthogonal to what we're trying to do here.

I agree it's better to return the partial response with the results for what we could successfully execute. My point is that this is orthogonal to what we're trying to do here.

I also think it would be good to have a partial response, but it's not a deal breaker for me. If the team is OK with the behavior I outlined here #3660 (comment) then I'm on board. It could also be fixed in a follow up PR to return a partial response.

I ended up going with

Running the protocol on each statement one-by-one (re running validation on distributed statements)

We could also re-run validation for logged statements as part of execute. So DistributingExecutor would just run the whole protocol (init txn, validate, log cmd, commit). It's a bit wasteful to re-run validation, but I think this is the simplest way to get this PR unblocked.

I think we're breaking an abstraction by passing the validator into the executor and it'll cause coupling problems down the line. It makes sense to run the validate-execute-commit one by one but not this way. Instead, the transaction logic should be pushed up to the top level that does the validate-execute loop instead of pushing it down into the execute.

Yeah doing the validation in DistributedExecutor was just meant to be an immediate-term thing so we can get this change merged without blocking on deciding how we really want the validation to be done. I'm personally OK with validating-executing each statement one by one. This is the way it was done at first. At some point @big-andy-coates changed it to validate everything up-front. It would be good to get his perspective on why this was done.

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

stevenpyzhang · 2019-11-01T21:13:13Z

Updates to PR:

New Configs

//for durability of command topic
ksql.internal.topic.replicas=3 //passed in through configs
ksql.internal.topic.min.insync.replicas=2 //passed in through configs
unclean.leader.election.enable=false

producer.acks=all // only for the command topic producer

//to enable transactions
isolation.level="read_commited" //consumer
transactional.id=[ksql_server_id] //producer

Added new configs ^
Transactional Producer waiting for commandStore to consume up to latest offset when beginning a transaction. This will only unblock if the commandRunner has processed all of the records up to the latest offset.
Don't create a TransactionalProducer if the incoming request only contains non-distributed statements (pull queries, list topics, ...etc). This will get addressed in a follow up PR.
ExceptionHandling for the transaction - a single transaction is handled inside DistributingExecutor.execute() which is wrapped in a try catch. Any exceptions will bubble up and be caught in the KsqlResource.handleTransactionalProduce() functions, where the transaction will then be aborted.
Changed protocol from a batched transaction to single statement transactions

The new protocol is:

	Create and initialize TransactionalProducer
	Validate all the statements at once
	for each statement:
		if non distributed, execute
		if distributed
			Start transaction
			Wait for CommandRunner consumer to process all records present after starting the transaction
			Validate the statement
			Send (returns QueuedCommandStatus) and Commit the record
			Try and Wait forQueueCommandStatus
		
		If exception while executing a statement, return error message specific to that statement.

One UX point is that some early statements may be committed, but a later one fails to be committed (since statements aren't batch committed to the command topic). We'd return an error message for that failed statement, but not any partial success responses

ksql> CREATE TABLE foo100(age BIGINT) WITH (KAFKA_TOPIC='foo',  VALUE_FORMAT='JSON');
>CREATE TABLE foo200(age BIGINT) WITH (KAFKA_TOPIC='foo',  VALUE_FORMAT='JSON');
>Show Tables;
>CREATE TABLE foo300(age BIGINT) WITH (KAFKA_TOPIC='foo',  VALUE_FORMAT='JSON');
>CREATE TABLE foo400(age BIGINT) WITH (KAFKA_TOPIC='foo',  VALUE_FORMAT='JSON'); // fails to be committed
>CREATE TABLE foo500(age BIGINT) WITH (KAFKA_TOPIC='foo',  VALUE_FORMAT='JSON');
>CREATE TABLE foo600(age BIGINT) WITH (KAFKA_TOPIC='foo',  VALUE_FORMAT='JSON');
>Show Tables;
Could not write the statement 'CREATE TABLE foo400(age BIGINT) WITH (KAFKA_TOPIC='foo',  VALUE_FORMAT='JSON');' into the command topic: Some exception
Caused by: Some exception
ksql> show tables;

 Table Name | Kafka Topic | Format | Windowed 
----------------------------------------------
 FOO100     | foo         | JSON   | false    
 FOO200     | foo         | JSON   | false    
 FOO300     | foo         | JSON   | false    
----------------------------------------------

This would need to be addressed in a follow up PR to return partial success responses.

…face to default kafka producer one

…log stream statemen

JimGalasyn

LGTM!

)" This reverts commit cba2877.

This reverts commit cba2877.

…entinc#3660)"" This reverts commit 219852b.

stevenpyzhang requested a review from a team as a code owner October 24, 2019 00:37

stevenpyzhang commented Oct 24, 2019

View reviewed changes

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

stevenpyzhang commented Oct 24, 2019

View reviewed changes

stevenpyzhang force-pushed the new-prototype-transactions branch from ce57314 to ba7c7a2 Compare October 24, 2019 00:42

purplefox reviewed Oct 24, 2019

View reviewed changes

agavra self-requested a review October 24, 2019 19:03

stevenpyzhang commented Oct 24, 2019

View reviewed changes

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

purplefox reviewed Oct 25, 2019

View reviewed changes

stevenpyzhang force-pushed the new-prototype-transactions branch 2 times, most recently from 5d0a69f to 81037c5 Compare October 28, 2019 18:35

vinothchandar reviewed Oct 28, 2019

View reviewed changes

stevenpyzhang force-pushed the new-prototype-transactions branch from 81037c5 to 8b6fef4 Compare October 28, 2019 21:43

rodesai reviewed Oct 29, 2019

View reviewed changes

stevenpyzhang force-pushed the new-prototype-transactions branch 7 times, most recently from 67467f5 to db2dd8a Compare November 1, 2019 20:38

stevenpyzhang added 9 commits November 15, 2019 09:22

process statements one at a time in rest server

7fd6cb2

more tests

ef3d03f

address Almog comments

b897c58

remove dependency on CommandRunner from TransactionalProducer

1396a09

documentation

495df9a

move transactional producer to distributing executor and change inter…

2567cae

…face to default kafka producer one

improved error handling and changed condition for sending processing …

f4a121f

…log stream statemen

rohan's comments

1117960

revert to using vanilla Kafka producer

5e42500

stevenpyzhang force-pushed the new-prototype-transactions branch from aad8942 to 5e42500 Compare November 15, 2019 17:41

stevenpyzhang requested a review from JimGalasyn as a code owner November 15, 2019 21:20

stevenpyzhang force-pushed the new-prototype-transactions branch 5 times, most recently from 2062bf8 to e862fb2 Compare November 15, 2019 23:06

update acls documentation

62756fd

stevenpyzhang force-pushed the new-prototype-transactions branch from e862fb2 to 62756fd Compare November 15, 2019 23:09

JimGalasyn approved these changes Nov 15, 2019

View reviewed changes

stevenpyzhang merged commit cba2877 into confluentinc:master Nov 16, 2019

This was referenced Nov 18, 2019

Consecutive calls to register topic should fail before enqueueing #2460

Closed

refactor: change processing log stream create to use KsqlResource #3899

Merged

agavra added a commit to agavra/ksql that referenced this pull request Nov 19, 2019

Revert "feat: Transactional Produces to Command Topic (confluentinc#3660

53ebc50

)" This reverts commit cba2877.

agavra added a commit that referenced this pull request Nov 19, 2019

Revert "feat: Transactional Produces to Command Topic (#3660)"

219852b

This reverts commit cba2877.

stevenpyzhang added a commit to stevenpyzhang/ksql that referenced this pull request Nov 21, 2019

Revert "Revert "feat: Transactional Produces to Command Topic (conflu…

ac005d9

…entinc#3660)"" This reverts commit 219852b.

This was referenced Nov 23, 2019

feat: never skip command in CommandRunner thread #3962

Merged

Update ksql configure-acls command and improve test coverage confluentinc/cli#338

Merged

agavra mentioned this pull request Dec 3, 2019

Be more rigorous about ensuring the right statements get executed. #2435

Closed

vcrfxia mentioned this pull request Jan 22, 2020

Rest APIs should wait for previous commands to complete by default. #2268

Closed

stevenpyzhang mentioned this pull request Feb 24, 2020

docs: add to security docs about transaction ACLs #4617

Merged

2 tasks

feat: Transactional Produces to Command Topic #3660

feat: Transactional Produces to Command Topic #3660

Conversation

stevenpyzhang commented Oct 24, 2019 • edited Loading

Description

Testing done

Reviewer checklist

Choose a reason for hiding this comment

purplefox left a comment

Choose a reason for hiding this comment

stevenpyzhang commented Oct 24, 2019 • edited Loading

purplefox commented Oct 24, 2019 • edited Loading

purplefox 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rodesai commented Oct 27, 2019

vinothchandar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rodesai commented Oct 28, 2019

stevenpyzhang commented Oct 28, 2019

rodesai commented Oct 28, 2019

stevenpyzhang commented Oct 28, 2019 • edited Loading

stevenpyzhang commented Oct 28, 2019 • edited Loading

rodesai commented Oct 28, 2019

Choose a reason for hiding this comment

stevenpyzhang Nov 1, 2019 • 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

Choose a reason for hiding this comment

stevenpyzhang Nov 1, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rodesai Nov 1, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stevenpyzhang Nov 1, 2019 • 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

stevenpyzhang commented Nov 1, 2019 • edited Loading

JimGalasyn left a comment

Choose a reason for hiding this comment

stevenpyzhang commented Oct 24, 2019 •

edited

Loading

stevenpyzhang commented Oct 24, 2019 •

edited

Loading

purplefox commented Oct 24, 2019 •

edited

Loading

stevenpyzhang commented Oct 28, 2019 •

edited

Loading

stevenpyzhang commented Oct 28, 2019 •

edited

Loading

stevenpyzhang Nov 1, 2019 •

edited

Loading

stevenpyzhang Nov 1, 2019 •

edited

Loading

rodesai Nov 1, 2019 •

edited

Loading

stevenpyzhang Nov 1, 2019 •

edited

Loading

stevenpyzhang commented Nov 1, 2019 •

edited

Loading