@Test public void testConsumerOverrides() { final Capture<Properties> consumerConfig = Capture.newInstance(); EasyMock.expect(consumerFactory.createConsumer(EasyMock.capture(consumerConfig))) .andReturn(consumer); EasyMock.replay(consumerFactory); consumerManager.createConsumer(groupName, ConsumerInstanceConfig.create(EmbeddedFormat.BINARY)); assertEquals("false", consumerConfig.getValue().get(ConsumerConfig.EXCLUDE_INTERNAL_TOPICS_CONFIG)); EasyMock.verify(consumerFactory); }
public String createConsumer(String group, ConsumerInstanceConfig instanceConfig) { String name = instanceConfig.getName(); if (instanceConfig.getId() != null) { name = instanceConfig.getId(); } if (name == null) { name = "rest-consumer-"; String serverId = this.config.getString(KafkaRestConfig.ID_CONFIG); if (!serverId.isEmpty()) { name += serverId + "-"; } name += UUID.randomUUID().toString(); } ConsumerInstanceId cid = new ConsumerInstanceId(group, name); synchronized (this) { if (consumers.containsKey(cid)) { throw Errors.consumerAlreadyExistsException(); } else { consumers.put(cid, null); } } boolean succeeded = false; try { log.debug("Creating consumer " + name + " in group " + group); Properties props = config.getConsumerProperties(); props.setProperty(KafkaRestConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers); props.setProperty(MAX_POLL_RECORDS_CONFIG, MAX_POLL_RECORDS_VALUE); props.setProperty("group.id", group); if (instanceConfig.getId() != null) { props.setProperty("consumer.id", instanceConfig.getId()); } if (instanceConfig.getAutoCommitEnable() != null) { props.setProperty("enable.auto.commit", instanceConfig.getAutoCommitEnable()); } if (instanceConfig.getAutoOffsetReset() != null) { props.setProperty("auto.offset.reset", instanceConfig.getAutoOffsetReset()); } props.setProperty("request.timeout.ms", "30000"); props.setProperty( "schema.registry.url", config.getString(KafkaRestConfig.SCHEMA_REGISTRY_URL_CONFIG) ); switch (instanceConfig.getFormat()) { case AVRO: props.put("key.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer"); break; case JSONSCHEMA: props.put("key.deserializer", "io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer"); break; case PROTOBUF: props.put("key.deserializer", "io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer"); break; case JSON: case BINARY: default: props.put( "key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer" ); props.put( "value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer" ); } Consumer consumer; try { if (consumerFactory == null) { consumer = new KafkaConsumer(props); } else { consumer = consumerFactory.createConsumer(props); } } catch (ConfigException e) { throw Errors.invalidConsumerConfigException(e.getMessage()); } KafkaConsumerState state = createConsumerState(instanceConfig, cid, consumer); synchronized (this) { consumers.put(cid, state); } succeeded = true; return name; } finally { if (!succeeded) { synchronized (this) { consumers.remove(cid); } } } }
KafkaConsumerManager { public String createConsumer(String group, ConsumerInstanceConfig instanceConfig) { String name = instanceConfig.getName(); if (instanceConfig.getId() != null) { name = instanceConfig.getId(); } if (name == null) { name = "rest-consumer-"; String serverId = this.config.getString(KafkaRestConfig.ID_CONFIG); if (!serverId.isEmpty()) { name += serverId + "-"; } name += UUID.randomUUID().toString(); } ConsumerInstanceId cid = new ConsumerInstanceId(group, name); synchronized (this) { if (consumers.containsKey(cid)) { throw Errors.consumerAlreadyExistsException(); } else { consumers.put(cid, null); } } boolean succeeded = false; try { log.debug("Creating consumer " + name + " in group " + group); Properties props = config.getConsumerProperties(); props.setProperty(KafkaRestConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers); props.setProperty(MAX_POLL_RECORDS_CONFIG, MAX_POLL_RECORDS_VALUE); props.setProperty("group.id", group); if (instanceConfig.getId() != null) { props.setProperty("consumer.id", instanceConfig.getId()); } if (instanceConfig.getAutoCommitEnable() != null) { props.setProperty("enable.auto.commit", instanceConfig.getAutoCommitEnable()); } if (instanceConfig.getAutoOffsetReset() != null) { props.setProperty("auto.offset.reset", instanceConfig.getAutoOffsetReset()); } props.setProperty("request.timeout.ms", "30000"); props.setProperty( "schema.registry.url", config.getString(KafkaRestConfig.SCHEMA_REGISTRY_URL_CONFIG) ); switch (instanceConfig.getFormat()) { case AVRO: props.put("key.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer"); break; case JSONSCHEMA: props.put("key.deserializer", "io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer"); break; case PROTOBUF: props.put("key.deserializer", "io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer"); break; case JSON: case BINARY: default: props.put( "key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer" ); props.put( "value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer" ); } Consumer consumer; try { if (consumerFactory == null) { consumer = new KafkaConsumer(props); } else { consumer = consumerFactory.createConsumer(props); } } catch (ConfigException e) { throw Errors.invalidConsumerConfigException(e.getMessage()); } KafkaConsumerState state = createConsumerState(instanceConfig, cid, consumer); synchronized (this) { consumers.put(cid, state); } succeeded = true; return name; } finally { if (!succeeded) { synchronized (this) { consumers.remove(cid); } } } } }
KafkaConsumerManager { public String createConsumer(String group, ConsumerInstanceConfig instanceConfig) { String name = instanceConfig.getName(); if (instanceConfig.getId() != null) { name = instanceConfig.getId(); } if (name == null) { name = "rest-consumer-"; String serverId = this.config.getString(KafkaRestConfig.ID_CONFIG); if (!serverId.isEmpty()) { name += serverId + "-"; } name += UUID.randomUUID().toString(); } ConsumerInstanceId cid = new ConsumerInstanceId(group, name); synchronized (this) { if (consumers.containsKey(cid)) { throw Errors.consumerAlreadyExistsException(); } else { consumers.put(cid, null); } } boolean succeeded = false; try { log.debug("Creating consumer " + name + " in group " + group); Properties props = config.getConsumerProperties(); props.setProperty(KafkaRestConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers); props.setProperty(MAX_POLL_RECORDS_CONFIG, MAX_POLL_RECORDS_VALUE); props.setProperty("group.id", group); if (instanceConfig.getId() != null) { props.setProperty("consumer.id", instanceConfig.getId()); } if (instanceConfig.getAutoCommitEnable() != null) { props.setProperty("enable.auto.commit", instanceConfig.getAutoCommitEnable()); } if (instanceConfig.getAutoOffsetReset() != null) { props.setProperty("auto.offset.reset", instanceConfig.getAutoOffsetReset()); } props.setProperty("request.timeout.ms", "30000"); props.setProperty( "schema.registry.url", config.getString(KafkaRestConfig.SCHEMA_REGISTRY_URL_CONFIG) ); switch (instanceConfig.getFormat()) { case AVRO: props.put("key.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer"); break; case JSONSCHEMA: props.put("key.deserializer", "io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer"); break; case PROTOBUF: props.put("key.deserializer", "io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer"); break; case JSON: case BINARY: default: props.put( "key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer" ); props.put( "value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer" ); } Consumer consumer; try { if (consumerFactory == null) { consumer = new KafkaConsumer(props); } else { consumer = consumerFactory.createConsumer(props); } } catch (ConfigException e) { throw Errors.invalidConsumerConfigException(e.getMessage()); } KafkaConsumerState state = createConsumerState(instanceConfig, cid, consumer); synchronized (this) { consumers.put(cid, state); } succeeded = true; return name; } finally { if (!succeeded) { synchronized (this) { consumers.remove(cid); } } } } KafkaConsumerManager(final KafkaRestConfig config);   KafkaConsumerManager(KafkaRestConfig config, KafkaConsumerFactory consumerFactory); }
KafkaConsumerManager { public String createConsumer(String group, ConsumerInstanceConfig instanceConfig) { String name = instanceConfig.getName(); if (instanceConfig.getId() != null) { name = instanceConfig.getId(); } if (name == null) { name = "rest-consumer-"; String serverId = this.config.getString(KafkaRestConfig.ID_CONFIG); if (!serverId.isEmpty()) { name += serverId + "-"; } name += UUID.randomUUID().toString(); } ConsumerInstanceId cid = new ConsumerInstanceId(group, name); synchronized (this) { if (consumers.containsKey(cid)) { throw Errors.consumerAlreadyExistsException(); } else { consumers.put(cid, null); } } boolean succeeded = false; try { log.debug("Creating consumer " + name + " in group " + group); Properties props = config.getConsumerProperties(); props.setProperty(KafkaRestConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers); props.setProperty(MAX_POLL_RECORDS_CONFIG, MAX_POLL_RECORDS_VALUE); props.setProperty("group.id", group); if (instanceConfig.getId() != null) { props.setProperty("consumer.id", instanceConfig.getId()); } if (instanceConfig.getAutoCommitEnable() != null) { props.setProperty("enable.auto.commit", instanceConfig.getAutoCommitEnable()); } if (instanceConfig.getAutoOffsetReset() != null) { props.setProperty("auto.offset.reset", instanceConfig.getAutoOffsetReset()); } props.setProperty("request.timeout.ms", "30000"); props.setProperty( "schema.registry.url", config.getString(KafkaRestConfig.SCHEMA_REGISTRY_URL_CONFIG) ); switch (instanceConfig.getFormat()) { case AVRO: props.put("key.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer"); break; case JSONSCHEMA: props.put("key.deserializer", "io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer"); break; case PROTOBUF: props.put("key.deserializer", "io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer"); break; case JSON: case BINARY: default: props.put( "key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer" ); props.put( "value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer" ); } Consumer consumer; try { if (consumerFactory == null) { consumer = new KafkaConsumer(props); } else { consumer = consumerFactory.createConsumer(props); } } catch (ConfigException e) { throw Errors.invalidConsumerConfigException(e.getMessage()); } KafkaConsumerState state = createConsumerState(instanceConfig, cid, consumer); synchronized (this) { consumers.put(cid, state); } succeeded = true; return name; } finally { if (!succeeded) { synchronized (this) { consumers.remove(cid); } } } } KafkaConsumerManager(final KafkaRestConfig config);   KafkaConsumerManager(KafkaRestConfig config, KafkaConsumerFactory consumerFactory); String createConsumer(String group, ConsumerInstanceConfig instanceConfig); void readRecords(
      final String group,
      final String instance,
      Class<? extends KafkaConsumerState<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>>
          consumerStateType,
      final long timeout,
      final long maxBytes,
      final ConsumerReadCallback<ClientKeyT, ClientValueT> callback
  ); Future commitOffsets(
      String group, String instance, final String async,
      final ConsumerOffsetCommitRequest offsetCommitRequest, final CommitCallback callback
  ); ConsumerCommittedResponse committed(
      String group,
      String instance,
      ConsumerCommittedRequest request
  ); long getBeginningOffset(String topic, int partition); long getEndOffset(String topic, int partition); Optional<Long> getOffsetForTime(
      String topic, int partition, Instant timestamp); void deleteConsumer(String group, String instance); void subscribe(String group, String instance, ConsumerSubscriptionRecord subscription); void unsubscribe(String group, String instance); ConsumerSubscriptionResponse subscription(String group, String instance); void seekToBeginning(String group, String instance, ConsumerSeekToRequest seekToRequest); void seekToEnd(String group, String instance, ConsumerSeekToRequest seekToRequest); void seekToOffset(
      String group,
      String instance,
      ConsumerSeekToOffsetRequest seekToOffsetRequest
  ); void assign(String group, String instance, ConsumerAssignmentRequest assignmentRequest); ConsumerAssignmentResponse assignment(String group, String instance); void shutdown(); }
KafkaConsumerManager { public String createConsumer(String group, ConsumerInstanceConfig instanceConfig) { String name = instanceConfig.getName(); if (instanceConfig.getId() != null) { name = instanceConfig.getId(); } if (name == null) { name = "rest-consumer-"; String serverId = this.config.getString(KafkaRestConfig.ID_CONFIG); if (!serverId.isEmpty()) { name += serverId + "-"; } name += UUID.randomUUID().toString(); } ConsumerInstanceId cid = new ConsumerInstanceId(group, name); synchronized (this) { if (consumers.containsKey(cid)) { throw Errors.consumerAlreadyExistsException(); } else { consumers.put(cid, null); } } boolean succeeded = false; try { log.debug("Creating consumer " + name + " in group " + group); Properties props = config.getConsumerProperties(); props.setProperty(KafkaRestConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers); props.setProperty(MAX_POLL_RECORDS_CONFIG, MAX_POLL_RECORDS_VALUE); props.setProperty("group.id", group); if (instanceConfig.getId() != null) { props.setProperty("consumer.id", instanceConfig.getId()); } if (instanceConfig.getAutoCommitEnable() != null) { props.setProperty("enable.auto.commit", instanceConfig.getAutoCommitEnable()); } if (instanceConfig.getAutoOffsetReset() != null) { props.setProperty("auto.offset.reset", instanceConfig.getAutoOffsetReset()); } props.setProperty("request.timeout.ms", "30000"); props.setProperty( "schema.registry.url", config.getString(KafkaRestConfig.SCHEMA_REGISTRY_URL_CONFIG) ); switch (instanceConfig.getFormat()) { case AVRO: props.put("key.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer"); break; case JSONSCHEMA: props.put("key.deserializer", "io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer"); break; case PROTOBUF: props.put("key.deserializer", "io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer"); props.put("value.deserializer", "io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer"); break; case JSON: case BINARY: default: props.put( "key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer" ); props.put( "value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer" ); } Consumer consumer; try { if (consumerFactory == null) { consumer = new KafkaConsumer(props); } else { consumer = consumerFactory.createConsumer(props); } } catch (ConfigException e) { throw Errors.invalidConsumerConfigException(e.getMessage()); } KafkaConsumerState state = createConsumerState(instanceConfig, cid, consumer); synchronized (this) { consumers.put(cid, state); } succeeded = true; return name; } finally { if (!succeeded) { synchronized (this) { consumers.remove(cid); } } } } KafkaConsumerManager(final KafkaRestConfig config);   KafkaConsumerManager(KafkaRestConfig config, KafkaConsumerFactory consumerFactory); String createConsumer(String group, ConsumerInstanceConfig instanceConfig); void readRecords(
      final String group,
      final String instance,
      Class<? extends KafkaConsumerState<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>>
          consumerStateType,
      final long timeout,
      final long maxBytes,
      final ConsumerReadCallback<ClientKeyT, ClientValueT> callback
  ); Future commitOffsets(
      String group, String instance, final String async,
      final ConsumerOffsetCommitRequest offsetCommitRequest, final CommitCallback callback
  ); ConsumerCommittedResponse committed(
      String group,
      String instance,
      ConsumerCommittedRequest request
  ); long getBeginningOffset(String topic, int partition); long getEndOffset(String topic, int partition); Optional<Long> getOffsetForTime(
      String topic, int partition, Instant timestamp); void deleteConsumer(String group, String instance); void subscribe(String group, String instance, ConsumerSubscriptionRecord subscription); void unsubscribe(String group, String instance); ConsumerSubscriptionResponse subscription(String group, String instance); void seekToBeginning(String group, String instance, ConsumerSeekToRequest seekToRequest); void seekToEnd(String group, String instance, ConsumerSeekToRequest seekToRequest); void seekToOffset(
      String group,
      String instance,
      ConsumerSeekToOffsetRequest seekToOffsetRequest
  ); void assign(String group, String instance, ConsumerAssignmentRequest assignmentRequest); ConsumerAssignmentResponse assignment(String group, String instance); void shutdown();  }