// refer https://github.com/IBM/sarama/issues/765#issuecomment-254333355
	sconfig.Producer.Retry.Max = 2
	sconfig.Producer.Retry.Backoff = (10 * time.Second)
	sconfig.Producer.Return.Successes = true
	sconfig.Producer.Return.Errors = true
	sconfig.Producer.RequiredAcks = 1
	sconfig.Producer.Timeout = (10 * time.Second)
	sconfig.Net.ReadTimeout = (10 * time.Second)
	sconfig.Net.DialTimeout = (10 * time.Second)

	}
	return target.isActive()
}

func (target *NSQTarget) isActive() (bool, error) {
	if target.producer == nil {
		producer, err := nsq.NewProducer(target.args.NSQDAddress.String(), target.config)
		if err != nil {
			return false, err
		}
		target.producer = producer
	}

	if err := target.producer.Ping(); err != nil {
		// To treat "connection refused" errors as errNotConnected.
		if xnet.IsConnRefusedErr(err) {

      assertEquals(0, Queues.drain(q, ImmutableList.of(), 1, 10, MILLISECONDS));

      Producer producer = new Producer(q, 1);
      // producing one, will ask for two
      Future<?> producerThread = threadPool.submit(producer);
      producer.beganProducing.await();

      // make sure we time out
      Stopwatch timer = Stopwatch.createStarted();

	config.Producer.Retry.Max = 2
	config.Producer.Retry.Backoff = (1 * time.Second)
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Producer.RequiredAcks = 1
	config.Producer.Timeout = (5 * time.Second)
	// Set Producer Compression
	cc, ok := codecs[strings.ToLower(args.Producer.Compression)]
	if ok {
		config.Producer.Compression = cc

		config.KV{
			Key:   target.KafkaSASLPassword,
			Value: cfg.SASL.Password,
		},
		config.KV{
			Key:   target.KafkaCompressionCodec,
			Value: cfg.Producer.Compression,
		},
		config.KV{
			Key:   target.KafkaCompressionLevel,
			Value: strconv.Itoa(cfg.Producer.CompressionLevel),
		},
	}
	return nil
}

// SetNotifyAMQP - helper for config migration from older config.

        logger.info(content);
        assertTrue(content.contains("テスト"));
        assertEquals("Writer", extractData.getValues("Creator")[0]);
        assertEquals("OpenOffice.org 3.0", extractData.getValues("Producer")[0]);
        assertEquals("D:20090627222631+09'00'", extractData.getValues("CreationDate")[0]);
    }

    public void test_getText_pass() {

        for (int i = 0; i < codes.length; i++) {
            if (codes[i] == code) {
                return Optional.of(msgs[i]);
            }
        }
        return Optional.empty();
    }

    // Producer for code->message pairs covering all known constants (happy path)
    static Stream<Arguments> knownCodeMessagePairs() {
        return Stream.of(Arguments.of(WinError.ERROR_SUCCESS, "The operation completed successfully."),

 * inserted by producers and extracted by consumers. Once created, the capacity cannot be increased.
 * Attempts to {@code put} an element into a full queue will result in the operation blocking;
 * attempts to {@code take} an element from an empty queue will similarly block.
 *
 * <p>This class supports an optional fairness policy for ordering waiting producer and consumer

		compressionCodecEnv := target.EnvKafkaProducerCompressionCodec
		if k != config.Default {
			compressionCodecEnv = compressionCodecEnv + config.Default + k
		}
		kafkaArgs.Producer.Compression = env.Get(compressionCodecEnv, kv.Get(target.KafkaCompressionCodec))

		compressionLevelEnv := target.EnvKafkaProducerCompressionLevel
		if k != config.Default {

MINIO_NOTIFY_KAFKA_PRODUCER_COMPRESSION_CODEC  (none|snappy|gzip|lz4|zstd) compression codec for producer messages
MINIO_NOTIFY_KAFKA_PRODUCER_COMPRESSION_LEVEL  (number)                    compression level for producer messages, defaults to '0'
```

To update the configuration, use `mc admin config get` command to get the current configuration.

```sh