For each checkpoint we create new {@link FlinkKafkaProducer} so that new transactions will not clash
with transactions created during previous checkpoints ({@code producer.initTransactions()} assures that we
obtain new producerId and epoch counters).
552607
private FlinkKafkaProducer<byte[], byte[]> createTransactionalProducer() throws FlinkKafka011Exception {
		String transactionalId = availableTransactionalIds.poll();
		if (transactionalId == null) {
			throw new FlinkKafka011Exception(
				FlinkKafka011ErrorCode.PRODUCERS_POOL_EMPTY,
				"Too many ongoing snapshots. Increase kafka producers pool size or decrease number of concurrent checkpoints.");
		}
		FlinkKafkaProducer<byte[], byte[]> producer = initTransactionalProducer(transactionalId, true);
		producer.initTransactions();
		return producer;
	}
815061
public final ControlNotFlushed createNewControlNotFlushed() throws MessageCreateFailedException {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(tc, "createNewControlNotFlushed");
    ControlNotFlushed msg = null;
    try {
      msg = new ControlNotFlushedImpl(MfpConstants.CONSTRUCTOR_NO_OP);
    }
    catch (MessageDecodeFailedException e) {
      /* No need to FFDC this as JsMsgObject will already have done so */
      // No FFDC code needed
      throw new MessageCreateFailedException(e);
    }
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "createNewControlNotFlushed");
    return msg;
  }
1193398
public void initNewList(int maxNumberOfTerms) throws IOException {
    if (closed) {
      result = null;
      closed = false;
    }
    if (segmentRegistration != null) {
      throw new IOException("missing segment name");
    } else {
      initNewListBasic(maxNumberOfTerms);
    }
  }
1244582
public static TransactionEdit createAborted(long writePointer, TransactionType type, long[] checkpointPointers) {
    return new TransactionEdit(writePointer, 0L, State.ABORTED, 0L, null, 0L, false, type, null, 0L, 0L,
        checkpointPointers);
  }
1342336
private void init(){

		errorProducerEnabled = MoskitoConfigurationHolder.getConfiguration().getBuiltinProducersConfig().isErrorProducer();

		statsMap = new ConcurrentHashMap<>();
        statsList = new CopyOnWriteArrayList<>();

		cumulatedStats = new ErrorStats("cumulated");
		statsList.add(cumulatedStats);

		catchers = new ConcurrentHashMap<>();

		//it is complicated to actually remove the error producer from the system if disabled, but the flag renders it invisible and
		//disables most methos.
		if (!errorProducerEnabled)
			return;

		ProducerRegistryFactory.getProducerRegistryInstance().registerProducer(this);


		//add charts for cumulated errors.
		AfterStartTasks.submitTask(new Runnable() {
			@Override
			public void run() {
				AccumulatorRepository.getInstance().createAccumulator(createAccumulatorDefinition("Errors.Cumulated.Total", "total", "cumulated"));
			}
		});
		AfterStartTasks.submitTask(new Runnable() {
			@Override
			public void run() {
				AccumulatorRepository.getInstance().createAccumulator(createAccumulatorDefinition("Errors.Cumulated.Initial", "initial", "cumulated"));
			}
		});

		errorHandlingConfig = null;
	}
1232055
public TraceStep startStep(String call, IStatsProducer producer, String methodName){
		TraceStep last = current;
		current = new TraceStep(call, producer, methodName);
		//it actually happened in production, we are still investigating why, but this fix should at least prevent the NPE for now.
		if (last!=null)
			last.addChild(current);
		return current;
	}
1294441
public CreateTransactionResult createTransaction(
            final long connectionId,
            final long txnId,
            final long uniqueId,
            final StoredProcedureInvocation invocation,
            final boolean isReadOnly,
            final boolean isSinglePartition,
            final boolean isEveryPartition,
            final int partition,
            final int messageSize,
            long nowNanos,
            final boolean isForReplay)
    {
        return m_dispatcher.createTransaction(
                connectionId,
                txnId,
                uniqueId,
                invocation,
                isReadOnly,
                isSinglePartition,
                isEveryPartition,
                new int [] { partition },
                messageSize,
                nowNanos,
                isForReplay);
    }
1265933
public void detachAllProducersForNewInputHandler(ProducerInputHandler newHandler)throws SIException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "detachAllProducersForNewInputHandler", newHandler);

    //detach all producers and update
    synchronized (_producers)
    {
      Iterator<MessageProducer> it = _producers.iterator();
      while(it.hasNext())
      {
        ProducerSessionImpl prod = (ProducerSessionImpl)it.next();
        it.remove();
        //attach this to the new input handler
        newHandler.attachProducer(prod);
        prod.updateInputHandler(newHandler);
      }
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "detachAllProducersForNewInputHandler");
  }
1167173
private SynchronizationAdapter createSynchronizationAdapter(Transaction transaction) {
      SynchronizationAdapter adapter = new SynchronizationAdapter(transaction, RemoteXid.create(uuid));
      try {
         transaction.registerSynchronization(adapter);
      } catch (RollbackException | SystemException e) {
         throw new CacheException(e);
      }
      if (trace) {
         log.tracef("Registered synchronization for transaction %s. Sync=%s", transaction, adapter);
      }
      return adapter;
   }
524272
private void emitRecordWithTimestampAndPeriodicWatermark(
			T record, KafkaTopicPartitionState<KPH> partitionState, long offset, long kafkaEventTimestamp) {
		@SuppressWarnings("unchecked")
		final KafkaTopicPartitionStateWithPeriodicWatermarks<T, KPH> withWatermarksState =
				(KafkaTopicPartitionStateWithPeriodicWatermarks<T, KPH>) partitionState;

		// extract timestamp - this accesses/modifies the per-partition state inside the
		// watermark generator instance, so we need to lock the access on the
		// partition state. concurrent access can happen from the periodic emitter
		final long timestamp;
		//noinspection SynchronizationOnLocalVariableOrMethodParameter
		synchronized (withWatermarksState) {
			timestamp = withWatermarksState.getTimestampForRecord(record, kafkaEventTimestamp);
		}

		// emit the record with timestamp, using the usual checkpoint lock to guarantee
		// atomicity of record emission and offset state update
		synchronized (checkpointLock) {
			sourceContext.collectWithTimestamp(record, timestamp);
			partitionState.setOffset(offset);
		}
	}
748448
def _create_producer(self, settings):
        """"""
        try:
            brokers = settings['KAFKA_HOSTS']
            self.logger.debug("Creating new kafka producer using brokers: " +
                               str(brokers))

            return KafkaProducer(bootstrap_servers=brokers,
                                 value_serializer=lambda m: json.dumps(m),
                                 retries=3,
                                 linger_ms=settings['KAFKA_PRODUCER_BATCH_LINGER_MS'],
                                 buffer_memory=settings['KAFKA_PRODUCER_BUFFER_BYTES'])
        except KeyError as e:
            self.logger.error('Missing setting named ' + str(e),
                               {'ex': traceback.format_exc()})
        except:
            self.logger.error("Couldn't initialize kafka producer in plugin.",
                               {'ex': traceback.format_exc()})
            raise
3123
@Override
	protected void flush() {
		// The Kafka 0.8 producer doesn't support flushing, we wait here
		// until all pending records are confirmed
		synchronized (pendingRecordsLock) {
			while (pendingRecords > 0) {
				try {
					pendingRecordsLock.wait();
				} catch (InterruptedException e) {
					// this can be interrupted when the Task has been cancelled.
					// by throwing an exception, we ensure that this checkpoint doesn't get confirmed
					throw new RuntimeException("Flushing got interrupted while checkpointing", e);
				}
			}
		}
	}
755025
public DisruptorQueue<ID, DATA> init() throws Exception {
        /* single producer "seems" to offer better performance */
        ringBuffer = RingBuffer.createSingleProducer(EVENT_FACTORY, ringSize);
        // ringBuffer = RingBuffer.createMultiProducer(EVENT_FACTORY, ringSize);

        if (!isEphemeralDisabled()) {
            int ephemeralBoundSize = Math.max(0, getEphemeralMaxSize());
            ephemeralStorage = new ConcurrentHashMap<>(
                    ephemeralBoundSize > 0 ? Math.min(ephemeralBoundSize, ringSize) : ringSize);
        }

        consumedSeq = new Sequence();
        ringBuffer.addGatingSequences(consumedSeq);
        long cursor = ringBuffer.getCursor();
        consumedSeq.set(cursor);
        knownPublishedSeq = cursor;

        super.init();

        return this;
    }
400361
private Producer<CloseableReference<CloseableImage>> newBitmapCacheGetToLocalTransformSequence(
      Producer<EncodedImage> inputProducer) {
    ThumbnailProducer<EncodedImage>[] defaultThumbnailProducers = new ThumbnailProducer[1];
    defaultThumbnailProducers[0] = mProducerFactory.newLocalExifThumbnailProducer();
    return newBitmapCacheGetToLocalTransformSequence(inputProducer, defaultThumbnailProducers);
  }
414554
private void createAccumulators(final OnDemandStatsProducer<S> producer, final Class producerClass, AccumulateWithSubClasses... annotations) {
		for (final AccumulateWithSubClasses annotation : annotations)
			if (annotation != null)
				AccumulatorUtil.getInstance(producerClass).createAccumulator(producer, annotation);
	}
393668
@Override
	public CompletableFuture<Acknowledge> triggerCheckpoint(
			ExecutionAttemptID executionAttemptID,
			long checkpointId,
			long checkpointTimestamp,
			CheckpointOptions checkpointOptions,
			boolean advanceToEndOfEventTime) {
		log.debug("Trigger checkpoint {}@{} for {}.", checkpointId, checkpointTimestamp, executionAttemptID);

		final CheckpointType checkpointType = checkpointOptions.getCheckpointType();
		if (advanceToEndOfEventTime && !(checkpointType.isSynchronous() && checkpointType.isSavepoint())) {
			throw new IllegalArgumentException("Only synchronous savepoints are allowed to advance the watermark to MAX.");
		}

		final Task task = taskSlotTable.getTask(executionAttemptID);

		if (task != null) {
			task.triggerCheckpointBarrier(checkpointId, checkpointTimestamp, checkpointOptions, advanceToEndOfEventTime);

			return CompletableFuture.completedFuture(Acknowledge.get());
		} else {
			final String message = "TaskManager received a checkpoint request for unknown task " + executionAttemptID + '.';

			log.debug(message);
			return FutureUtils.completedExceptionally(new CheckpointException(message));
		}
	}
192291
public static void initialiseLinks(TraceCompletionInformation ci, long fragmentBaseTime, Node n,
            StringBuilder nodeId) {
        // Add Communication to represent a potential 'CausedBy' link from one or more fragments back to
        // this node
        TraceCompletionInformation.Communication c = new TraceCompletionInformation.Communication();

        c.getIds().add(nodeId.toString());

        // Define a a multi-consumer as potentially multiple CausedBy correlations may be created
        // back to this node
        c.setMultipleConsumers(true);

        // Calculate the base duration for the communication
        c.setBaseDuration(n.getTimestamp() - fragmentBaseTime);

        c.setExpire(System.currentTimeMillis()+
                TraceCompletionInformation.Communication.DEFAULT_EXPIRY_WINDOW_MILLIS);

        if (log.isLoggable(Level.FINEST)) {
            log.finest("Adding communication to completion information: ci=" + ci + " comms=" + c);
        }

        ci.getCommunications().add(c);

        if (n.getClass() == Producer.class) {
            // Get correlation ids
            List<CorrelationIdentifier> cids = n.findCorrelationIds(Scope.Interaction, Scope.ControlFlow);

            if (!cids.isEmpty()) {
                c = new TraceCompletionInformation.Communication();

                for (int i = 0; i < cids.size(); i++) {
                    c.getIds().add(cids.get(i).getValue());
                }

                c.setMultipleConsumers(((Producer) n).multipleConsumers());

                // Calculate the base duration for the communication
                c.setBaseDuration(n.getTimestamp() - fragmentBaseTime);

                c.setExpire(System.currentTimeMillis() +
                        TraceCompletionInformation.Communication.DEFAULT_EXPIRY_WINDOW_MILLIS);

                if (log.isLoggable(Level.FINEST)) {
                    log.finest("Adding communication to completion information: ci=" + ci + " comms=" + c);
                }

                ci.getCommunications().add(c);
            }
        } else if (n.containerNode()) {
            ContainerNode cn = (ContainerNode) n;
            for (int i = 0; i < cn.getNodes().size(); i++) {
                int len = nodeId.length();
                nodeId.append(':');
                nodeId.append(i);
                initialiseLinks(ci, fragmentBaseTime, cn.getNodes().get(i), nodeId);

                // Remove this child's specific path, so that next iteration will add a different path number
                nodeId.delete(len, nodeId.length());
            }
        }
    }