content: /*
   * Copyright 2021-2024 Aklivity Inc.
   *
   * Aklivity licenses this file to you under the Apache License,
   * version 2.0 (the "License"); you may not use this file except in compliance
   * with the License. You may obtain a copy of the License at:
   *
   *   http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   * License for the specific language governing permissions and limitations
   * under the License.
   */
  package io.aklivity.zilla.runtime.binding.kafka.internal.stream;
  
  import static io.aklivity.zilla.runtime.binding.kafka.internal.types.ProxyAddressProtocol.STREAM;
  import static io.aklivity.zilla.runtime.engine.budget.BudgetCreditor.NO_BUDGET_ID;
  import static io.aklivity.zilla.runtime.engine.budget.BudgetDebitor.NO_DEBITOR_INDEX;
  import static io.aklivity.zilla.runtime.engine.buffer.BufferPool.NO_SLOT;
  
  import java.util.List;
  import java.util.function.Consumer;
  import java.util.function.LongFunction;
  import java.util.function.UnaryOperator;
  
  import org.agrona.DirectBuffer;
  import org.agrona.MutableDirectBuffer;
  import org.agrona.collections.LongLongConsumer;
  import org.agrona.concurrent.UnsafeBuffer;
  
  import io.aklivity.zilla.runtime.binding.kafka.config.KafkaSaslConfig;
  import io.aklivity.zilla.runtime.binding.kafka.config.KafkaServerConfig;
  import io.aklivity.zilla.runtime.binding.kafka.internal.KafkaBinding;
  import io.aklivity.zilla.runtime.binding.kafka.internal.KafkaConfiguration;
  import io.aklivity.zilla.runtime.binding.kafka.internal.config.KafkaBindingConfig;
  import io.aklivity.zilla.runtime.binding.kafka.internal.config.KafkaRouteConfig;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.Flyweight;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.OctetsFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.codec.RequestHeaderFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.codec.ResponseHeaderFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.codec.produce.InitProducerIdRequestFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.codec.produce.InitProducerIdResponseFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.AbortFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.BeginFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.DataFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.EndFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.ExtensionFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.KafkaBeginExFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.KafkaInitProducerIdBeginExFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.KafkaResetExFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.ProxyBeginExFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.ResetFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.SignalFW;
  import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.WindowFW;
  import io.aklivity.zilla.runtime.engine.EngineContext;
  import io.aklivity.zilla.runtime.engine.binding.BindingHandler;
  import io.aklivity.zilla.runtime.engine.binding.function.MessageConsumer;
  import io.aklivity.zilla.runtime.engine.budget.BudgetDebitor;
  import io.aklivity.zilla.runtime.engine.buffer.BufferPool;
  import io.aklivity.zilla.runtime.engine.concurrent.Signaler;
  
  public final class KafkaClientInitProducerIdFactory extends KafkaClientSaslHandshaker implements BindingHandler
  {
      private static final int ERROR_NONE = 0;
  
      private static final int SIGNAL_NEXT_REQUEST = 1;
  
      private static final DirectBuffer EMPTY_BUFFER = new UnsafeBuffer();
      private static final OctetsFW EMPTY_OCTETS = new OctetsFW().wrap(EMPTY_BUFFER, 0, 0);
      private static final Consumer<OctetsFW.Builder> EMPTY_EXTENSION = ex -> {};
  
      private static final short INIT_PRODUCE_ID_API_KEY = 22;
      private static final short INIT_PRODUCE_ID_API_VERSION = 1;
  
      private final BeginFW beginRO = new BeginFW();
      private final DataFW dataRO = new DataFW();
      private final EndFW endRO = new EndFW();
      private final AbortFW abortRO = new AbortFW();
      private final ResetFW resetRO = new ResetFW();
      private final WindowFW windowRO = new WindowFW();
      private final SignalFW signalRO = new SignalFW();
      private final ExtensionFW extensionRO = new ExtensionFW();
      private final KafkaBeginExFW kafkaBeginExRO = new KafkaBeginExFW();
  
      private final BeginFW.Builder beginRW = new BeginFW.Builder();
      private final DataFW.Builder dataRW = new DataFW.Builder();
      private final EndFW.Builder endRW = new EndFW.Builder();
      private final AbortFW.Builder abortRW = new AbortFW.Builder();
      private final ResetFW.Builder resetRW = new ResetFW.Builder();
      private final WindowFW.Builder windowRW = new WindowFW.Builder();
      private final KafkaBeginExFW.Builder kafkaBeginExRW = new KafkaBeginExFW.Builder();
      private final KafkaResetExFW.Builder kafkaResetExRW = new KafkaResetExFW.Builder();
      private final ProxyBeginExFW.Builder proxyBeginExRW = new ProxyBeginExFW.Builder();
  
      private final RequestHeaderFW.Builder requestHeaderRW = new RequestHeaderFW.Builder();
      private final InitProducerIdRequestFW.Builder initProducerIdRequestRW = new InitProducerIdRequestFW.Builder();
  
      private final ResponseHeaderFW responseHeaderRO = new ResponseHeaderFW();
      private final InitProducerIdResponseFW initProducerrIdResponseRO = new InitProducerIdResponseFW();
  
      private final KafkaInitProducerIdClientDecoder decodeSaslHandshakeResponse = this::decodeSaslHandshakeResponse;
      private final KafkaInitProducerIdClientDecoder decodeSaslHandshake = this::decodeSaslHandshake;
      private final KafkaInitProducerIdClientDecoder decodeSaslHandshakeMechanisms = this::decodeSaslHandshakeMechanisms;
      private final KafkaInitProducerIdClientDecoder decodeSaslHandshakeMechanism = this::decodeSaslHandshakeMechanism;
      private final KafkaInitProducerIdClientDecoder decodeSaslAuthenticateResponse = this::decodeSaslAuthenticateResponse;
      private final KafkaInitProducerIdClientDecoder decodeSaslAuthenticate = this::decodeSaslAuthenticate;
      private final KafkaInitProducerIdClientDecoder decodeInitProducerIdResponse = this::decodeInitProducerIdResponse;
  
      private final KafkaInitProducerIdClientDecoder decodeIgnoreAll = this::decodeIgnoreAll;
      private final KafkaInitProducerIdClientDecoder decodeReject = this::decodeReject;
  
      private final int kafkaTypeId;
      private final int proxyTypeId;
      private final MutableDirectBuffer writeBuffer;
      private final MutableDirectBuffer extBuffer;
      private final BufferPool decodePool;
      private final BufferPool encodePool;
      private final Signaler signaler;
      private final BindingHandler streamFactory;
      private final UnaryOperator<KafkaSaslConfig> resolveSasl;
      private final LongFunction<KafkaBindingConfig> supplyBinding;
      private final LongFunction<BudgetDebitor> supplyDebitor;
  
      public KafkaClientInitProducerIdFactory(
          KafkaConfiguration config,
          EngineContext context,
          LongFunction<KafkaBindingConfig> supplyBinding,
          LongFunction<BudgetDebitor> supplyDebitor,
          Signaler signaler,
          BindingHandler streamFactory,
          UnaryOperator<KafkaSaslConfig> resolveSasl)
      {
          super(config, context);
          this.kafkaTypeId = context.supplyTypeId(KafkaBinding.NAME);
          this.proxyTypeId = context.supplyTypeId("proxy");
          this.signaler = signaler;
          this.streamFactory = streamFactory;
          this.resolveSasl = resolveSasl;
          this.writeBuffer = new UnsafeBuffer(new byte[context.writeBuffer().capacity()]);
          this.extBuffer = new UnsafeBuffer(new byte[context.writeBuffer().capacity()]);
          this.decodePool = context.bufferPool();
          this.encodePool = context.bufferPool();
          this.supplyBinding = supplyBinding;
          this.supplyDebitor = supplyDebitor;
      }
  
      @Override
      public MessageConsumer newStream(
          int msgTypeId,
          DirectBuffer buffer,
          int index,
          int length,
          MessageConsumer application)
      {
          final BeginFW begin = beginRO.wrap(buffer, index, index + length);
          final long originId = begin.originId();
          final long routedId = begin.routedId();
          final long initialId = begin.streamId();
          final long affinity = begin.affinity();
          final long authorization = begin.authorization();
          final OctetsFW extension = begin.extension();
          final ExtensionFW beginEx = extensionRO.tryWrap(extension.buffer(), extension.offset(), extension.limit());
          final KafkaBeginExFW kafkaBeginEx = beginEx != null && beginEx.typeId() == kafkaTypeId ?
              kafkaBeginExRO.tryWrap(extension.buffer(), extension.offset(), extension.limit()) : null;
  
          assert kafkaBeginEx.kind() == KafkaBeginExFW.KIND_INIT_PRODUCER_ID;
          final KafkaInitProducerIdBeginExFW initProducerIdBeginEx = kafkaBeginEx.initProducerId();
          final long producerId = initProducerIdBeginEx.producerId();
          final short producerEpoch = initProducerIdBeginEx.producerEpoch();
  
          MessageConsumer newStream = null;
  
          final KafkaBindingConfig binding = supplyBinding.apply(routedId);
          final KafkaRouteConfig resolved = binding != null ?
              binding.resolve(authorization, null, null) : null;
  
          if (resolved != null)
          {
              final long resolvedId = resolved.id;
              final List<KafkaServerConfig> servers = binding.servers();
              final KafkaSaslConfig sasl = resolveSasl.apply(binding.sasl());
  
              newStream = new KafkaInitProducerIdStream(
                      application,
                      originId,
                      routedId,
                      initialId,
                      affinity,
                      resolvedId,
                      producerId,
                      producerEpoch,
                      servers,
                      sasl)::onApplication;
          }
  
          return newStream;
      }
  
      private MessageConsumer newStream(
          MessageConsumer sender,
          long originId,
          long routedId,
          long streamId,
          long sequence,
          long acknowledge,
          int maximum,
          long traceId,
          long authorization,
          long affinity,
          Consumer<OctetsFW.Builder> extension)
      {
          final BeginFW begin = beginRW.wrap(writeBuffer, 0, writeBuffer.capacity())
                  .originId(originId)
                  .routedId(routedId)
                  .streamId(streamId)
                  .sequence(sequence)
                  .acknowledge(acknowledge)
                  .maximum(maximum)
                  .traceId(traceId)
                  .authorization(authorization)
                  .affinity(affinity)
                  .extension(extension)
                  .build();
  
          final MessageConsumer receiver =
                  streamFactory.newStream(begin.typeId(), begin.buffer(), begin.offset(), begin.sizeof(), sender);
  
          receiver.accept(begin.typeId(), begin.buffer(), begin.offset(), begin.sizeof());
  
          return receiver;
      }
  
      private void doBegin(
          MessageConsumer receiver,
          long originId,
          long routedId,
          long streamId,
          long sequence,
          long acknowledge,
          int maximum,
          long traceId,
          long authorization,
          long affinity,
          Consumer<OctetsFW.Builder> extension)
      {
          final BeginFW begin = beginRW.wrap(writeBuffer, 0, writeBuffer.capacity())
                  .originId(originId)
                  .routedId(routedId)
                  .streamId(streamId)
                  .sequence(sequence)
                  .acknowledge(acknowledge)
                  .maximum(maximum)
                  .traceId(traceId)
                  .authorization(authorization)
                  .affinity(affinity)
                  .extension(extension)
                  .build();
  
          receiver.accept(begin.typeId(), begin.buffer(), begin.offset(), begin.sizeof());
      }
  
      private void doData(
          MessageConsumer receiver,
          long originId,
          long routedId,
          long streamId,
          long sequence,
          long acknowledge,
          int maximum,
          long traceId,
          long authorization,
          long budgetId,
          int reserved,
          DirectBuffer payload,
          int offset,
          int length,
          Consumer<OctetsFW.Builder> extension)
      {
          final DataFW data = dataRW.wrap(writeBuffer, 0, writeBuffer.capacity())
                  .originId(originId)
                  .routedId(routedId)
                  .streamId(streamId)
                  .sequence(sequence)
                  .acknowledge(acknowledge)
                  .maximum(maximum)
                  .traceId(traceId)
                  .authorization(authorization)
                  .budgetId(budgetId)
                  .reserved(reserved)
                  .payload(payload, offset, length)
                  .extension(extension)
                  .build();
  
          receiver.accept(data.typeId(), data.buffer(), data.offset(), data.sizeof());
      }
  
      private void doData(
          MessageConsumer receiver,
          long originId,
          long routedId,
          long streamId,
          long sequence,
          long acknowledge,
          int maximum,
          long traceId,
          long authorization,
          long budgetId,
          int reserved,
          DirectBuffer payload,
          int offset,
          int length,
          Flyweight extension)
      {
          final DataFW data = dataRW.wrap(writeBuffer, 0, writeBuffer.capacity())
              .originId(originId)
              .routedId(routedId)
              .streamId(streamId)
              .sequence(sequence)
              .acknowledge(acknowledge)
              .maximum(maximum)
              .traceId(traceId)
              .authorization(authorization)
              .budgetId(budgetId)
              .reserved(reserved)
              .payload(payload, offset, length)
              .extension(extension.buffer(), extension.offset(), extension.sizeof())
              .build();
  
          receiver.accept(data.typeId(), data.buffer(), data.offset(), data.sizeof());
      }
  
      private void doEnd(
          MessageConsumer receiver,
          long originId,
          long routedId,
          long streamId,
          long sequence,
          long acknowledge,
          int maximum,
          long traceId,
          long authorization,
          Consumer<OctetsFW.Builder> extension)
      {
          final EndFW end = endRW.wrap(writeBuffer, 0, writeBuffer.capacity())
                  .originId(originId)
                  .routedId(routedId)
                  .streamId(streamId)
                  .sequence(sequence)
                  .acknowledge(acknowledge)
                  .maximum(maximum)
                  .traceId(traceId)
                  .authorization(authorization)
                  .extension(extension)
                  .build();
  
          receiver.accept(end.typeId(), end.buffer(), end.offset(), end.sizeof());
      }
  
      private void doAbort(
          MessageConsumer receiver,
          long originId,
          long routedId,
          long streamId,
          long sequence,
          long acknowledge,
          int maximum,
          long traceId,
          long authorization,
          Consumer<OctetsFW.Builder> extension)
      {
          final AbortFW abort = abortRW.wrap(writeBuffer, 0, writeBuffer.capacity())
                  .originId(originId)
                  .routedId(routedId)
                  .streamId(streamId)
                  .sequence(sequence)
                  .acknowledge(acknowledge)
                  .maximum(maximum)
                  .traceId(traceId)
                  .authorization(authorization)
                  .extension(extension)
                  .build();
  
          receiver.accept(abort.typeId(), abort.buffer(), abort.offset(), abort.sizeof());
      }
  
      private void doWindow(
          MessageConsumer sender,
          long originId,
          long routedId,
          long streamId,
          long sequence,
          long acknowledge,
          int maximum,
          long traceId,
          long authorization,
          long budgetId,
          int padding)
      {
          final WindowFW window = windowRW.wrap(writeBuffer, 0, writeBuffer.capacity())
                  .originId(originId)
                  .routedId(routedId)
                  .streamId(streamId)
                  .sequence(sequence)
                  .acknowledge(acknowledge)
                  .maximum(maximum)
                  .traceId(traceId)
                  .authorization(authorization)
                  .budgetId(budgetId)
                  .padding(padding)
                  .build();
  
          sender.accept(window.typeId(), window.buffer(), window.offset(), window.sizeof());
      }
  
      private void doReset(
          MessageConsumer sender,
          long originId,
          long routedId,
          long streamId,
          long sequence,
          long acknowledge,
          int maximum,
          long traceId,
          long authorization,
          Flyweight extension)
      {
          final ResetFW reset = resetRW.wrap(writeBuffer, 0, writeBuffer.capacity())
                  .originId(originId)
                  .routedId(routedId)
                  .streamId(streamId)
                  .sequence(sequence)
                  .acknowledge(acknowledge)
                  .maximum(maximum)
                  .traceId(traceId)
                  .authorization(authorization)
                  .extension(extension.buffer(), extension.offset(), extension.sizeof())
                  .build();
  
          sender.accept(reset.typeId(), reset.buffer(), reset.offset(), reset.sizeof());
      }
  
      @FunctionalInterface
      private interface KafkaInitProducerIdClientDecoder
      {
          int decode(
              KafkaInitProducerIdClient client,
              long traceId,
              long authorization,
              long budgetId,
              int reserved,
              MutableDirectBuffer buffer,
              int offset,
              int progress,
              int limit);
      }
  
      private int decodeInitProducerIdResponse(
          KafkaInitProducerIdClient client,
          long traceId,
          long authorization,
          long budgetId,
          int reserved,
          DirectBuffer buffer,
          int offset,
          int progress,
          int limit)
      {
          final int length = limit - progress;
  
          decode:
          if (length != 0)
          {
              final ResponseHeaderFW responseHeader = responseHeaderRO.tryWrap(buffer, progress, limit);
              if (responseHeader == null)
              {
                  break decode;
              }
  
              progress = responseHeader.limit();
  
  
              final InitProducerIdResponseFW initProducerIdResponse = initProducerrIdResponseRO.tryWrap(buffer, progress, limit);
              if (initProducerIdResponse == null)
              {
                  break decode;
              }
  
              progress = initProducerIdResponse.limit();
  
              short errorCode = initProducerIdResponse.errorCode();
              if (errorCode == ERROR_NONE)
              {
                  client.onDecodeInitProducerrIdResponse(
                      traceId, initProducerIdResponse.producerId(), initProducerIdResponse.producerEpoch());
              }
              else
              {
                  client.errorCode = errorCode;
                  client.decoder = decodeReject;
              }
          }
  
          return progress;
      }
  
  
      private int decodeReject(
          KafkaInitProducerIdClient client,
          long traceId,
          long authorization,
          long budgetId,
          int reserved,
          DirectBuffer buffer,
          int offset,
          int progress,
          int limit)
      {
          client.cleanupNetwork(traceId);
          client.decoder = decodeIgnoreAll;
          return limit;
      }
  
      private int decodeIgnoreAll(
          KafkaInitProducerIdClient client,
          long traceId,
          long authorization,
          long budgetId,
          int reserved,
          DirectBuffer buffer,
          int offset,
          int progress,
          int limit)
      {
          return limit;
      }
  
      private final class KafkaInitProducerIdStream
      {
          private final MessageConsumer application;
          private final long originId;
          private final long routedId;
          private final long initialId;
          private final long replyId;
          private final long affinity;
          private final KafkaInitProducerIdClient client;
  
          private int state;
  
          private long initialSeq;
          private long initialAck;
          private int initialMax;
  
          private long replySeq;
          private long replyAck;
          private int replyMax;
          private int replyPad;
  
          private long replyBudgetId;
  
          KafkaInitProducerIdStream(
              MessageConsumer application,
              long originId,
              long routedId,
              long initialId,
              long affinity,
              long resolvedId,
              long producerId,
              short producerEpoch,
              List<KafkaServerConfig> servers,
              KafkaSaslConfig sasl)
          {
              this.application = application;
              this.originId = originId;
              this.routedId = routedId;
              this.initialId = initialId;
              this.replyId = supplyReplyId.applyAsLong(initialId);
              this.affinity = affinity;
              this.client = new KafkaInitProducerIdClient(this, routedId, resolvedId, producerId, producerEpoch, servers, sasl);
          }
  
          private void onApplication(
              int msgTypeId,
              DirectBuffer buffer,
              int index,
              int length)
          {
              switch (msgTypeId)
              {
              case BeginFW.TYPE_ID:
                  final BeginFW begin = beginRO.wrap(buffer, index, index + length);
                  onApplicationBegin(begin);
                  break;
              case DataFW.TYPE_ID:
                  final DataFW data = dataRO.wrap(buffer, index, index + length);
                  onApplicationData(data);
                  break;
              case EndFW.TYPE_ID:
                  final EndFW end = endRO.wrap(buffer, index, index + length);
                  onApplicationEnd(end);
                  break;
              case AbortFW.TYPE_ID:
                  final AbortFW abort = abortRO.wrap(buffer, index, index + length);
                  onApplicationAbort(abort);
                  break;
              case WindowFW.TYPE_ID:
                  final WindowFW window = windowRO.wrap(buffer, index, index + length);
                  onApplicationWindow(window);
                  break;
              case ResetFW.TYPE_ID:
                  final ResetFW reset = resetRO.wrap(buffer, index, index + length);
                  onApplicationReset(reset);
                  break;
              default:
                  break;
              }
          }
  
          private void onApplicationBegin(
              BeginFW begin)
          {
              final long traceId = begin.traceId();
              final long authorization = begin.authorization();
  
              state = KafkaState.openingInitial(state);
  
              client.doNetworkBegin(traceId, authorization, affinity);
  
              doApplicationWindow(traceId,  0L, 0, 0, 0);
          }
  
          private void onApplicationData(
              DataFW data)
          {
              final long traceId = data.traceId();
  
              client.cleanupNetwork(traceId);
          }
  
          private void onApplicationEnd(
              EndFW end)
          {
              final long traceId = end.traceId();
              final long authorization = end.authorization();
  
              state = KafkaState.closedInitial(state);
  
              client.doNetworkEnd(traceId, authorization);
          }
  
          private void onApplicationAbort(
              AbortFW abort)
          {
              final long traceId = abort.traceId();
  
              state = KafkaState.closedInitial(state);
  
              client.doNetworkAbortIfNecessary(traceId);
          }
  
          private void onApplicationWindow(
              WindowFW window)
          {
              final long sequence = window.sequence();
              final long acknowledge = window.acknowledge();
              final int maximum = window.maximum();
              final long budgetId = window.budgetId();
              final int padding = window.padding();
  
              assert acknowledge <= sequence;
              assert sequence <= replySeq;
              assert acknowledge >= replyAck;
              assert maximum >= replyMax;
  
              this.replyAck = acknowledge;
              this.replyMax = maximum;
              this.replyPad = padding;
              this.replyBudgetId = budgetId;
  
              assert replyAck <= replySeq;
          }
  
          private void onApplicationReset(
              ResetFW reset)
          {
              final long traceId = reset.traceId();
  
              state = KafkaState.closedReply(state);
  
              client.doNetworkResetIfNecessary(traceId);
          }
  
          private boolean isApplicationReplyOpen()
          {
              return KafkaState.replyOpening(state);
          }
  
          private void doApplicationBegin(
              long traceId,
              long authorization,
              Consumer<OctetsFW.Builder> extension)
          {
              state = KafkaState.openingReply(state);
  
              doBegin(application, originId, routedId, replyId, replySeq, replyAck, replyMax,
                      traceId, authorization, affinity, extension);
          }
  
          private void doApplicationEnd(
              long traceId)
          {
              state = KafkaState.closedReply(state);
              doEnd(application, originId, routedId, replyId, replySeq, replyAck, replyMax,
                      traceId, client.authorization, EMPTY_EXTENSION);
          }
  
          private void doApplicationAbort(
              long traceId)
          {
              state = KafkaState.closedReply(state);
              doAbort(application, originId, routedId, replyId, replySeq, replyAck, replyMax,
                      traceId, client.authorization, EMPTY_EXTENSION);
          }
  
          private void doApplicationWindow(
              long traceId,
              long budgetId,
              int minInitialNoAck,
              int minInitialPad,
              int minInitialMax)
          {
              final long newInitialAck = Math.max(initialSeq - minInitialNoAck, initialAck);
  
              if (newInitialAck > initialAck || minInitialMax > initialMax || !KafkaState.initialOpened(state))
              {
                  initialAck = newInitialAck;
                  assert initialAck <= initialSeq;
  
                  initialMax = minInitialMax;
  
                  state = KafkaState.openedInitial(state);
  
                  doWindow(application, originId, routedId, initialId, initialSeq, initialAck, initialMax,
                          traceId, client.authorization, budgetId, minInitialPad);
              }
          }
  
          private void doApplicationReset(
              long traceId,
              Flyweight extension)
          {
              state = KafkaState.closedInitial(state);
  
              doReset(application, originId, routedId, initialId, initialSeq, initialAck, initialMax,
                      traceId, client.authorization, extension);
          }
  
          private void doApplicationAbortIfNecessary(
              long traceId)
          {
              if (KafkaState.replyOpening(state) && !KafkaState.replyClosed(state))
              {
                  doApplicationAbort(traceId);
              }
          }
  
          private void doApplicationResetIfNecessary(
              long traceId,
              Flyweight extension)
          {
              if (KafkaState.initialOpening(state) && !KafkaState.initialClosed(state))
              {
                  doApplicationReset(traceId, extension);
              }
          }
  
          private void cleanupApplication(
              long traceId,
              int error)
          {
              final KafkaResetExFW kafkaResetEx = kafkaResetExRW.wrap(extBuffer, 0, extBuffer.capacity())
                  .typeId(kafkaTypeId)
                  .error(error)
                  .build();
  
              cleanupApplication(traceId, kafkaResetEx);
          }
  
          private void cleanupApplication(
              long traceId,
              Flyweight extension)
          {
              doApplicationResetIfNecessary(traceId, extension);
              doApplicationAbortIfNecessary(traceId);
          }
      }
  
      private final class KafkaInitProducerIdClient extends KafkaSaslClient
      {
          private final LongLongConsumer encodeSaslHandshakeRequest = this::doEncodeSaslHandshakeRequest;
          private final LongLongConsumer encodeSaslAuthenticateRequest = this::doEncodeSaslAuthenticateRequest;
          private final LongLongConsumer encodeInitProducerIdRequest = this::doEncodeInitProducerIdRequest;
  
          private final KafkaInitProducerIdStream delegate;
          private final long producerId;
          private final short producerEpoch;
  
          private short errorCode;
  
          private MessageConsumer network;
          private int state;
          private long authorization;
  
          private long initialSeq;
          private long initialAck;
          private int initialMax;
          private int initialMin;
          private int initialPad;
          private long initialBudgetId = NO_BUDGET_ID;
          private long initialDebIndex = NO_DEBITOR_INDEX;
  
          private long replySeq;
          private long replyAck;
          private int replyMax;
  
          private int encodeSlot = NO_SLOT;
          private int encodeSlotOffset;
          private long encodeSlotTraceId;
  
          private int decodeSlot = NO_SLOT;
          private int decodeSlotOffset;
          private int decodeSlotReserved;
  
          private int nextResponseId;
  
  
          private BudgetDebitor initialDeb;
          private KafkaInitProducerIdClientDecoder decoder;
          private LongLongConsumer encoder;
  
          KafkaInitProducerIdClient(
              KafkaInitProducerIdStream delegate,
              long originId,
              long routedId,
              long producerId,
              short producerEpoch,
              List<KafkaServerConfig> servers,
              KafkaSaslConfig sasl)
          {
              super(servers, sasl, originId, routedId);
              this.delegate = delegate;
              this.producerId = producerId;
              this.producerEpoch = producerEpoch;
              this.encoder = sasl != null ? encodeSaslHandshakeRequest : encodeInitProducerIdRequest;
  
              this.decoder = decodeReject;
          }
  
          private void onNetwork(
              int msgTypeId,
              DirectBuffer buffer,
              int index,
              int length)
          {
              switch (msgTypeId)
              {
              case BeginFW.TYPE_ID:
                  final BeginFW begin = beginRO.wrap(buffer, index, index + length);
                  onNetworkBegin(begin);
                  break;
              case DataFW.TYPE_ID:
                  final DataFW data = dataRO.wrap(buffer, index, index + length);
                  onNetworkData(data);
                  break;
              case EndFW.TYPE_ID:
                  final EndFW end = endRO.wrap(buffer, index, index + length);
                  onNetworkEnd(end);
                  break;
              case AbortFW.TYPE_ID:
                  final AbortFW abort = abortRO.wrap(buffer, index, index + length);
                  onNetworkAbort(abort);
                  break;
              case ResetFW.TYPE_ID:
                  final ResetFW reset = resetRO.wrap(buffer, index, index + length);
                  onNetworkReset(reset);
                  break;
              case WindowFW.TYPE_ID:
                  final WindowFW window = windowRO.wrap(buffer, index, index + length);
                  onNetworkWindow(window);
                  break;
              case SignalFW.TYPE_ID:
                  final SignalFW signal = signalRO.wrap(buffer, index, index + length);
                  onNetworkSignal(signal);
                  break;
              default:
                  break;
              }
          }
  
          private void onNetworkBegin(
              BeginFW begin)
          {
              final long traceId = begin.traceId();
  
              authorization = begin.authorization();
              state = KafkaState.openingReply(state);
  
              doNetworkWindow(traceId, 0L, 0, 0, decodePool.slotCapacity());
          }
  
          private void onNetworkData(
              DataFW data)
          {
              final long sequence = data.sequence();
              final long acknowledge = data.acknowledge();
              final long traceId = data.traceId();
              final long budgetId = data.budgetId();
  
              assert acknowledge <= sequence;
              assert sequence >= replySeq;
  
              replySeq = sequence + data.reserved();
              authorization = data.authorization();
  
              assert replyAck <= replySeq;
  
              if (replySeq > replyAck + replyMax)
              {
                  cleanupNetwork(traceId);
              }
              else
              {
                  if (decodeSlot == NO_SLOT)
                  {
                      decodeSlot = decodePool.acquire(initialId);
                  }
  
                  if (decodeSlot == NO_SLOT)
                  {
                      cleanupNetwork(traceId);
                  }
                  else
                  {
                      final OctetsFW payload = data.payload();
                      int reserved = data.reserved();
                      int offset = payload.offset();
                      int limit = payload.limit();
  
                      final MutableDirectBuffer buffer = decodePool.buffer(decodeSlot);
                      buffer.putBytes(decodeSlotOffset, payload.buffer(), offset, limit - offset);
                      decodeSlotOffset += limit - offset;
                      decodeSlotReserved += reserved;
  
                      offset = 0;
                      limit = decodeSlotOffset;
                      reserved = decodeSlotReserved;
  
                      decodeNetwork(traceId, authorization, budgetId, reserved, buffer, offset, limit);
                  }
              }
          }
  
          private void onNetworkEnd(
              EndFW end)
          {
              final long traceId = end.traceId();
  
              state = KafkaState.closedReply(state);
  
              cleanupDecodeSlotIfNecessary();
  
              if (!delegate.isApplicationReplyOpen())
              {
                  cleanupNetwork(traceId);
              }
              else if (decodeSlot == NO_SLOT)
              {
                  delegate.doApplicationEnd(traceId);
              }
          }
  
          private void onNetworkAbort(
              AbortFW abort)
          {
              final long traceId = abort.traceId();
  
              state = KafkaState.closedReply(state);
  
              cleanupNetwork(traceId);
          }
  
          private void onNetworkReset(
              ResetFW reset)
          {
              final long traceId = reset.traceId();
  
              state = KafkaState.closedInitial(state);
  
              cleanupNetwork(traceId);
          }
  
          private void onNetworkWindow(
              WindowFW window)
          {
              final long sequence = window.sequence();
              final long acknowledge = window.acknowledge();
              final int minimum = window.minimum();
              final int maximum = window.maximum();
              final long traceId = window.traceId();
              final long budgetId = window.budgetId();
              final int padding = window.padding();
  
              assert acknowledge <= sequence;
              assert sequence <= initialSeq;
              assert acknowledge >= initialAck;
              assert maximum + acknowledge >= initialMax + initialAck;
  
              this.initialAck = acknowledge;
              this.initialMax = maximum;
              this.initialPad = padding;
              this.initialMin = minimum;
              this.initialBudgetId = budgetId;
  
              assert initialAck <= initialSeq;
  
              this.authorization = window.authorization();
  
              state = KafkaState.openedInitial(state);
  
              if (initialBudgetId != NO_BUDGET_ID && initialDebIndex == NO_DEBITOR_INDEX)
              {
                  initialDeb = supplyDebitor.apply(initialBudgetId);
                  initialDebIndex = initialDeb.acquire(initialBudgetId, initialId, this::doNetworkDataIfNecessary);
                  assert initialDebIndex != NO_DEBITOR_INDEX;
              }
  
              doNetworkDataIfNecessary(budgetId);
  
              doEncodeRequestIfNecessary(traceId, budgetId);
          }
  
          private void doNetworkDataIfNecessary(
              long traceId)
          {
              if (encodeSlot != NO_SLOT)
              {
                  final MutableDirectBuffer buffer = encodePool.buffer(encodeSlot);
                  final int limit = encodeSlotOffset;
  
                  encodeNetwork(traceId, authorization, initialBudgetId, buffer, 0, limit);
              }
          }
  
          private void onNetworkSignal(
              SignalFW signal)
          {
              final long traceId = signal.traceId();
              final int signalId = signal.signalId();
  
              if (signalId == SIGNAL_NEXT_REQUEST)
              {
                  doEncodeRequestIfNecessary(traceId, initialBudgetId);
              }
          }
  
          private void doNetworkBegin(
              long traceId,
              long authorization,
              long affinity)
          {
              state = KafkaState.openingInitial(state);
  
              Consumer<OctetsFW.Builder> extension = EMPTY_EXTENSION;
  
              if (server != null)
              {
                  extension = e -> e.set((b, o, l) -> proxyBeginExRW.wrap(b, o, l)
                                                                    .typeId(proxyTypeId)
                                                                    .address(a -> a.inet(i -> i.protocol(p -> p.set(STREAM))
                                                                                               .source("0.0.0.0")
                                                                                               .destination(server.host)
                                                                                               .sourcePort(0)
                                                                                               .destinationPort(server.port)))
                                                                    .infos(i -> i.item(ii -> ii.authority(server.host)))
                                                                    .build()
                                                                    .sizeof());
              }
  
              network = newStream(this::onNetwork, originId, routedId, initialId, initialSeq, initialAck, initialMax,
                      traceId, authorization, affinity, extension);
          }
  
          @Override
          protected void doNetworkData(
              long traceId,
              long budgetId,
              DirectBuffer buffer,
              int offset,
              int limit)
          {
              if (encodeSlot != NO_SLOT)
              {
                  final MutableDirectBuffer encodeBuffer = encodePool.buffer(encodeSlot);
                  encodeBuffer.putBytes(encodeSlotOffset, buffer, offset, limit - offset);
                  encodeSlotOffset += limit - offset;
                  encodeSlotTraceId = traceId;
  
                  buffer = encodeBuffer;
                  offset = 0;
                  limit = encodeSlotOffset;
              }
  
              encodeNetwork(traceId, authorization, budgetId, buffer, offset, limit);
          }
  
          private void doNetworkEnd(
              long traceId,
              long authorization)
          {
              state = KafkaState.closedInitial(state);
  
              cleanupEncodeSlotIfNecessary();
              cleanupBudgetIfNecessary();
  
              doEnd(network, originId, routedId, initialId, initialSeq, initialAck, initialMax,
                  traceId, authorization, EMPTY_EXTENSION);
          }
  
          private void doNetworkAbortIfNecessary(
              long traceId)
          {
              if (!KafkaState.initialClosed(state))
              {
                  doAbort(network, originId, routedId, initialId, initialSeq, initialAck, initialMax,
                      traceId, authorization, EMPTY_EXTENSION);
                  state = KafkaState.closedInitial(state);
              }
  
              cleanupEncodeSlotIfNecessary();
              cleanupBudgetIfNecessary();
          }
  
          private void doNetworkResetIfNecessary(
              long traceId)
          {
              if (!KafkaState.replyClosed(state))
              {
                  doReset(network, originId, routedId, replyId, replySeq, replyAck, replyMax,
                      traceId, authorization, EMPTY_OCTETS);
                  state = KafkaState.closedReply(state);
              }
  
              cleanupDecodeSlotIfNecessary();
          }
  
          private void doNetworkWindow(
              long traceId,
              long budgetId,
              int minReplyNoAck,
              int minReplyPad,
              int minReplyMax)
          {
              final long newReplyAck = Math.max(replySeq - minReplyNoAck, replyAck);
  
              if (newReplyAck > replyAck || minReplyMax > replyMax || !KafkaState.replyOpened(state))
              {
                  replyAck = newReplyAck;
                  assert replyAck <= replySeq;
  
                  replyMax = minReplyMax;
  
                  state = KafkaState.openedReply(state);
  
                  doWindow(network, originId, routedId, replyId, replySeq, replyAck, replyMax,
                      traceId, authorization, budgetId, minReplyPad);
              }
          }
  
          private void doEncodeRequestIfNecessary(
              long traceId,
              long budgetId)
          {
              if (nextRequestId == nextResponseId)
              {
                  encoder.accept(traceId, budgetId);
              }
          }
  
          private void doEncodeInitProducerIdRequest(
              long traceId,
              long budgetId)
          {
  
              final MutableDirectBuffer encodeBuffer = writeBuffer;
              final int encodeOffset = DataFW.FIELD_OFFSET_PAYLOAD;
              final int encodeLimit = encodeBuffer.capacity();
  
              int encodeProgress = encodeOffset;
  
              final RequestHeaderFW requestHeader = requestHeaderRW.wrap(encodeBuffer, encodeProgress, encodeLimit)
                  .length(0)
                  .apiKey(INIT_PRODUCE_ID_API_KEY)
                  .apiVersion(INIT_PRODUCE_ID_API_VERSION)
                  .correlationId(0)
                  .clientId(clientId)
                  .build();
  
              encodeProgress = requestHeader.limit();
  
              final InitProducerIdRequestFW initProducerIdRequest =
                  initProducerIdRequestRW.wrap(encodeBuffer, encodeProgress, encodeLimit)
                      .producerId(producerId)
                      .producerEpoch(producerEpoch)
                      .build();
  
              encodeProgress = initProducerIdRequest.limit();
  
              final int requestId = nextRequestId++;
              final int requestSize = encodeProgress - encodeOffset - RequestHeaderFW.FIELD_OFFSET_API_KEY;
  
              requestHeaderRW.wrap(encodeBuffer, requestHeader.offset(), requestHeader.limit())
                  .length(requestSize)
                  .apiKey(requestHeader.apiKey())
                  .apiVersion(requestHeader.apiVersion())
                  .correlationId(requestId)
                  .clientId(requestHeader.clientId())
                  .build();
  
              doNetworkData(traceId, budgetId, encodeBuffer, encodeOffset, encodeProgress);
  
              decoder = decodeInitProducerIdResponse;
          }
  
          private void encodeNetwork(
              long traceId,
              long authorization,
              long budgetId,
              DirectBuffer buffer,
              int offset,
              int limit)
          {
              final int length = limit - offset;
              final int initialBudget = Math.max(initialMax - (int)(initialSeq - initialAck), 0);
              final int reservedMax = Math.max(Math.min(length + initialPad, initialBudget), initialMin);
  
              int reserved = reservedMax;
  
              flush:
              if (reserved > 0)
              {
  
                  boolean claimed = false;
  
                  if (initialDebIndex != NO_DEBITOR_INDEX)
                  {
                      reserved = initialDeb.claim(traceId, initialDebIndex, initialId, reserved, reserved, 0);
                      claimed = reserved > 0;
                  }
  
                  if (reserved < initialPad || reserved == initialPad && length > 0)
                  {
                      break flush;
                  }
  
                  doData(network, originId, routedId, initialId, initialSeq, initialAck, initialMax,
                      traceId, authorization, budgetId, reserved, buffer, offset, length, EMPTY_EXTENSION);
  
                  initialSeq += reserved;
  
                  assert initialAck <= initialSeq;
              }
  
              final int flushed = Math.max(reserved - initialPad, 0);
              final int remaining = length - flushed;
              if (remaining > 0)
              {
                  if (encodeSlot == NO_SLOT)
                  {
                      encodeSlot = encodePool.acquire(initialId);
                  }
  
                  if (encodeSlot == NO_SLOT)
                  {
                      cleanupNetwork(traceId);
                  }
                  else
                  {
                      final MutableDirectBuffer encodeBuffer = encodePool.buffer(encodeSlot);
                      encodeBuffer.putBytes(0, buffer, offset + flushed, remaining);
                      encodeSlotOffset = remaining;
                  }
              }
              else
              {
                  cleanupEncodeSlotIfNecessary();
              }
          }
  
          private void decodeNetwork(
              long traceId,
              long authorization,
              long budgetId,
              int reserved,
              MutableDirectBuffer buffer,
              int offset,
              int limit)
          {
              KafkaInitProducerIdClientDecoder previous = null;
              int progress = offset;
              while (progress <= limit && previous != decoder)
              {
                  previous = decoder;
                  progress = decoder.decode(this, traceId, authorization, budgetId, reserved, buffer, offset, progress, limit);
              }
  
              if (progress < limit)
              {
                  if (decodeSlot == NO_SLOT)
                  {
                      decodeSlot = decodePool.acquire(initialId);
                  }
  
                  if (decodeSlot == NO_SLOT)
                  {
                      cleanupNetwork(traceId);
                  }
                  else
                  {
                      final MutableDirectBuffer decodeBuffer = decodePool.buffer(decodeSlot);
                      decodeBuffer.putBytes(0, buffer, progress, limit - progress);
                      decodeSlotOffset = limit - progress;
                      decodeSlotReserved = (limit - progress) * reserved / (limit - offset);
                  }
  
                  doNetworkWindow(traceId, budgetId, decodeSlotOffset, 0, replyMax);
              }
              else
              {
                  cleanupDecodeSlotIfNecessary();
  
                  if (KafkaState.replyClosing(state))
                  {
                      delegate.doApplicationEnd(traceId);
                  }
                  else if (reserved > 0)
                  {
                      doNetworkWindow(traceId, budgetId, 0, 0, replyMax);
                  }
              }
          }
  
          @Override
          protected void doDecodeSaslHandshakeResponse(
              long traceId)
          {
              decoder = decodeSaslHandshakeResponse;
          }
  
          @Override
          protected void doDecodeSaslHandshake(
              long traceId)
          {
              decoder = decodeSaslHandshake;
          }
  
          @Override
          protected void doDecodeSaslHandshakeMechanisms(
              long traceId)
          {
              decoder = decodeSaslHandshakeMechanisms;
          }
  
          @Override
          protected void doDecodeSaslHandshakeMechansim(
              long traceId)
          {
              decoder = decodeSaslHandshakeMechanism;
          }
  
          @Override
          protected void doDecodeSaslAuthenticateResponse(
              long traceId)
          {
              decoder = decodeSaslAuthenticateResponse;
          }
  
          @Override
          protected void doDecodeSaslAuthenticate(
              long traceId)
          {
              decoder = decodeSaslAuthenticate;
          }
  
          @Override
          protected void onDecodeSaslHandshakeResponse(
              long traceId,
              long authorization,
              int errorCode)
          {
              switch (errorCode)
              {
              case ERROR_NONE:
                  encoder = encodeSaslAuthenticateRequest;
                  decoder = decodeSaslAuthenticateResponse;
                  break;
              default:
                  delegate.cleanupApplication(traceId, errorCode);
                  doNetworkEnd(traceId, authorization);
                  break;
              }
          }
  
          @Override
          protected void onDecodeSaslAuthenticateResponse(
              long traceId,
              long authorization,
              int errorCode)
          {
              switch (errorCode)
              {
              case ERROR_NONE:
                  encoder = encodeInitProducerIdRequest;
                  decoder = decodeInitProducerIdResponse;
                  break;
              default:
                  delegate.cleanupApplication(traceId, errorCode);
                  doNetworkEnd(traceId, authorization);
                  break;
              }
          }
  
          @Override
          protected void onDecodeSaslResponse(
              long traceId)
          {
              nextResponseId++;
              signaler.signalNow(originId, routedId, initialId, traceId, SIGNAL_NEXT_REQUEST, 0);
          }
  
          private void onDecodeInitProducerrIdResponse(
              long traceId,
              long newProducerId,
              short newProducerEpoch)
          {
              delegate.doApplicationBegin(traceId, authorization,  ex -> ex.set((b, o, l) -> kafkaBeginExRW.wrap(b, o, l)
                  .typeId(kafkaTypeId)
                  .initProducerId(p -> p.producerId(newProducerId).producerEpoch(newProducerEpoch))
                  .build()
                  .sizeof()));
          }
  
          private void cleanupNetwork(
              long traceId)
          {
              doNetworkResetIfNecessary(traceId);
              doNetworkAbortIfNecessary(traceId);
  
              delegate.cleanupApplication(traceId, errorCode);
          }
  
          private void cleanupDecodeSlotIfNecessary()
          {
              if (decodeSlot != NO_SLOT)
              {
                  decodePool.release(decodeSlot);
                  decodeSlot = NO_SLOT;
                  decodeSlotOffset = 0;
                  decodeSlotReserved = 0;
              }
          }
  
          private void cleanupEncodeSlotIfNecessary()
          {
              if (encodeSlot != NO_SLOT)
              {
                  encodePool.release(encodeSlot);
                  encodeSlot = NO_SLOT;
                  encodeSlotOffset = 0;
                  encodeSlotTraceId = 0;
              }
          }
  
          private void cleanupBudgetIfNecessary()
          {
              if (initialDebIndex != NO_DEBITOR_INDEX)
              {
                  initialDeb.release(initialDebIndex, initialId);
                  initialDebIndex = NO_DEBITOR_INDEX;
              }
          }
      }
  
  }
  