Hi Shengyi, 1) Unfortunately no, see the related docs: https://kafka.apache. org/0110/documentation.html#upgrade_11_exactly_once_semantics
2) You cannot set the internal topic message format on the client side (producer, consumer, streams, ..) it is decided on the broker side only. 3) You can read more about the upgrade path on https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging#KIP-98-ExactlyOnceDeliveryandTransactionalMessaging-MessageFormat (section Compatibility, Deprecation, and Migration Plan) Guozhang On Wed, Apr 18, 2018 at 8:05 PM, sy.pan <shengyi....@gmail.com> wrote: > Hi, All: > > I have upgraded my Kafka cluster from 0.10.2 to 1.1 recently. After > rolling upgrade, the broker version related configuration is : > > inter.broker.protocol.version = 1.1 > log.message.format.version = 0.10.2 > > I keep the log message format as low version because not all clients could > upgrade in a short time. > When i test Kafka stream EOS feature, get the ERROR log: > > [ERROR][StreamThread]: stream-thread [xxx-b0b080fb-StreamThread-2] Failed > to close task manager due to the following error: > org.apache.kafka.common.KafkaException: Cannot execute transactional > method because we are in an error state > at org.apache.kafka.clients.producer.internals.TransactionManager. > maybeFailWithError(TransactionManager.java:784) > at org.apache.kafka.clients.producer.internals. > TransactionManager.beginAbort(TransactionManager.java:229) > at org.apache.kafka.clients.producer.KafkaProducer. > abortTransaction(KafkaProducer.java:660) > at org.apache.kafka.streams.processor.internals. > StreamTask.closeSuspended(StreamTask.java:486) > at org.apache.kafka.streams.processor.internals. > StreamTask.close(StreamTask.java:546) > at org.apache.kafka.streams.processor.internals. > AssignedTasks.close(AssignedTasks.java:405) > at org.apache.kafka.streams.processor.internals. > TaskManager.shutdown(TaskManager.java:260) > at org.apache.kafka.streams.processor.internals. > StreamThread.completeShutdown(StreamThread.java:1107) > at org.apache.kafka.streams.processor.internals. > StreamThread.run(StreamThread.java:731) > Caused by: > org.apache.kafka.common.errors.UnsupportedForMessageFormatException: > The message format version on the broker does not support the request. > > So my question is : > > 1) Can I use Kafka stream EOS when the log.message.format.version < 0.11 > > 2) How to set internal topic message format(created by Kafka stream) ? I > couldn’t find in Kafka Streams Configs > > 3) What the performance impact if set log.message.format.version = 1.1 > when some clients’ version is still 0.10.2 > (trigger message format transforming on broker side?) > > > > Thank you! -- -- Guozhang
