From d80e66d9b8466607c40ec009108f592901358507 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arnaud=20Jardin=C3=A9?= Date: Fri, 8 Feb 2019 11:41:31 +0100 Subject: [PATCH] Actuator health for Kafka-streams binder Add documentation about health indicator Fix failing tests + add tests for multiple Kafka streams Polishing Resolves #544 --- docs/src/main/asciidoc/kafka-streams.adoc | 28 ++ .../spring-cloud-stream-binder-kafka.adoc | 2 +- .../pom.xml | 5 + .../streams/KStreamBinderConfiguration.java | 3 +- .../KafkaStreamsBinderHealthIndicator.java | 81 +++++ ...amsBinderHealthIndicatorConfiguration.java | 42 +++ ...afkaStreamsBinderHealthIndicatorTests.java | 319 ++++++++++++++++++ 7 files changed, 478 insertions(+), 2 deletions(-) create mode 100644 spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java create mode 100644 spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicatorConfiguration.java create mode 100644 spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java diff --git a/docs/src/main/asciidoc/kafka-streams.adoc b/docs/src/main/asciidoc/kafka-streams.adoc index b41f4328f..ef3366cd5 100644 --- a/docs/src/main/asciidoc/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka-streams.adoc @@ -693,3 +693,31 @@ StreamsBuilderFactoryBean streamsBuilderFactoryBean = context.getBean("&stream-b By default, the `Kafkastreams.cleanup()` method is called when the binding is stopped. See https://docs.spring.io/spring-kafka/reference/html/_reference.html#_configuration[the Spring Kafka documentation]. To modify this behavior simply add a single `CleanupConfig` `@Bean` (configured to clean up on start, stop, or neither) to the application context; the bean will be detected and wired into the factory bean. + +=== Health Indicator + +The health indicator requires the dependency `spring-boot-starter-actuator`. For maven use: +[source,xml] +---- + + org.springframework.boot + spring-boot-starter-actuator + +---- + +Spring Cloud Stream Binder Kafka Streams provides a health indicator to check the state of the underlying Kafka threads. +Spring Cloud Stream defines a property `management.health.binders.enabled` to enable the health indicator. See the +https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_health_indicator[Spring Cloud Stream documentation]. + +The health indicator provides the following details for each Kafka threads: + +* Thread name +* Thread state: `CREATED`, `RUNNING`, `PARTITIONS_REVOKED`, `PARTITIONS_ASSIGNED`, `PENDING_SHUTDOWN` or `DEAD` +* Active tasks: task ID and partitions +* Standby tasks: task ID and partitions + +By default, only the global status is visible (`UP` or `DOWN`). To show the details, the property `management.endpoint.health.show-details` must be set to `ALWAYS` or `WHEN_AUTHORIZED`. +For more details about the health information, see the +https://docs.spring.io/spring-boot/docs/current/reference/html/production-ready-endpoints.html#production-ready-health[Spring Boot Actuator documentation]. + +NOTE: The status of the health indicator is `UP` if all the Kafka threads registered are in the `RUNNING` state. diff --git a/docs/src/main/asciidoc/spring-cloud-stream-binder-kafka.adoc b/docs/src/main/asciidoc/spring-cloud-stream-binder-kafka.adoc index 2343467ce..d574c8253 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream-binder-kafka.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream-binder-kafka.adoc @@ -10,7 +10,7 @@ [[spring-cloud-stream-binder-kafka-reference]] = Spring Cloud Stream Kafka Binder Reference Guide -Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein, Henryk Konsek, Gary Russell +Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein, Henryk Konsek, Gary Russell, Arnaud Jardiné :doctype: book :toc: :toclevels: 4 diff --git a/spring-cloud-stream-binder-kafka-streams/pom.xml b/spring-cloud-stream-binder-kafka-streams/pom.xml index d6f36c6e9..9281ecd9c 100644 --- a/spring-cloud-stream-binder-kafka-streams/pom.xml +++ b/spring-cloud-stream-binder-kafka-streams/pom.xml @@ -22,6 +22,11 @@ org.springframework.cloud spring-cloud-stream-binder-kafka-core + + org.springframework.boot + spring-boot-starter-actuator + true + org.springframework.boot spring-boot-configuration-processor diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java index fdc1c24f3..6961fbfa3 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java @@ -39,7 +39,8 @@ import org.springframework.context.annotation.Import; * @author Soby Chacko */ @Configuration -@Import({ KafkaAutoConfiguration.class }) +@Import({ KafkaAutoConfiguration.class, + KafkaStreamsBinderHealthIndicatorConfiguration.class }) public class KStreamBinderConfiguration { @Bean diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java new file mode 100644 index 000000000..5197a264f --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java @@ -0,0 +1,81 @@ +/* + * Copyright 2019-2019 the original author or authors. + * + * Licensed 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 org.springframework.cloud.stream.binder.kafka.streams; + +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.processor.TaskMetadata; +import org.apache.kafka.streams.processor.ThreadMetadata; + +import org.springframework.boot.actuate.health.AbstractHealthIndicator; +import org.springframework.boot.actuate.health.Health; +import org.springframework.boot.actuate.health.Status; + +/** + * Health indicator for Kafka Streams. + * + * @author Arnaud Jardiné + */ +class KafkaStreamsBinderHealthIndicator extends AbstractHealthIndicator { + + private final KafkaStreamsRegistry kafkaStreamsRegistry; + + KafkaStreamsBinderHealthIndicator(KafkaStreamsRegistry kafkaStreamsRegistry) { + super("Kafka-streams health check failed"); + this.kafkaStreamsRegistry = kafkaStreamsRegistry; + } + + @Override + protected void doHealthCheck(Health.Builder builder) throws Exception { + boolean up = true; + for (KafkaStreams kStream : kafkaStreamsRegistry.getKafkaStreams()) { + up &= kStream.state().isRunning(); + builder.withDetails(buildDetails(kStream)); + } + builder.status(up ? Status.UP : Status.DOWN); + } + + private static Map buildDetails(KafkaStreams kStreams) { + final Map details = new HashMap<>(); + if (kStreams.state().isRunning()) { + for (ThreadMetadata metadata : kStreams.localThreadsMetadata()) { + details.put("threadName", metadata.threadName()); + details.put("threadState", metadata.threadState()); + details.put("activeTasks", taskDetails(metadata.activeTasks())); + details.put("standbyTasks", taskDetails(metadata.standbyTasks())); + } + } + return details; + } + + private static Map taskDetails(Set taskMetadata) { + final Map details = new HashMap<>(); + for (TaskMetadata metadata : taskMetadata) { + details.put("taskId", metadata.taskId()); + details.put("partitions", + metadata.topicPartitions().stream().map( + p -> "partition=" + p.partition() + ", topic=" + p.topic()) + .collect(Collectors.toList())); + } + return details; + } + +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicatorConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicatorConfiguration.java new file mode 100644 index 000000000..9f83bc4a6 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicatorConfiguration.java @@ -0,0 +1,42 @@ +/* + * Copyright 2019-2019 the original author or authors. + * + * Licensed 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 org.springframework.cloud.stream.binder.kafka.streams; + +import org.springframework.boot.actuate.autoconfigure.health.ConditionalOnEnabledHealthIndicator; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * Configuration class for Kafka-streams binder health indicator beans. + * + * @author Arnaud Jardiné + */ +@Configuration +@ConditionalOnClass(name = "org.springframework.boot.actuate.health.HealthIndicator") +@ConditionalOnEnabledHealthIndicator("binders") +class KafkaStreamsBinderHealthIndicatorConfiguration { + + @Bean + @ConditionalOnBean(KafkaStreamsRegistry.class) + KafkaStreamsBinderHealthIndicator kafkaStreamsBinderHealthIndicator( + KafkaStreamsRegistry kafkaStreamsRegistry) { + return new KafkaStreamsBinderHealthIndicator(kafkaStreamsRegistry); + } + +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java new file mode 100644 index 000000000..f5fac096a --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java @@ -0,0 +1,319 @@ +/* + * Copyright 2019-2019 the original author or authors. + * + * Licensed 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 org.springframework.cloud.stream.binder.kafka.streams.integration; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.kstream.KStream; +import org.assertj.core.util.Lists; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.actuate.health.Health; +import org.springframework.boot.actuate.health.HealthIndicator; +import org.springframework.boot.actuate.health.Status; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.Output; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsApplicationSupportProperties; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.SendResult; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.util.concurrent.ListenableFuture; +import org.springframework.util.concurrent.ListenableFutureCallback; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Arnaud Jardiné + */ +public class KafkaStreamsBinderHealthIndicatorTests { + + @ClassRule + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, + "out", "out2"); + + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule + .getEmbeddedKafka(); + + @BeforeClass + public static void setUp() { + System.setProperty("logging.level.org.apache.kafka", "OFF"); + } + + @Test + public void healthIndicatorUpTest() throws Exception { + try (ConfigurableApplicationContext context = singleStream()) { + receive(context, + Lists.newArrayList(new ProducerRecord<>("in", "{\"id\":\"123\"}"), + new ProducerRecord<>("in", "{\"id\":\"123\"}")), + Status.UP, "out"); + } + } + + @Test + public void healthIndicatorDownTest() throws Exception { + try (ConfigurableApplicationContext context = singleStream()) { + receive(context, + Lists.newArrayList(new ProducerRecord<>("in", "{\"id\":\"123\"}"), + new ProducerRecord<>("in", "{\"id\":\"124\"}")), + Status.DOWN, "out"); + } + } + + @Test + public void healthIndicatorUpMultipleKStreamsTest() throws Exception { + try (ConfigurableApplicationContext context = multipleStream()) { + receive(context, + Lists.newArrayList(new ProducerRecord<>("in", "{\"id\":\"123\"}"), + new ProducerRecord<>("in2", "{\"id\":\"123\"}")), + Status.UP, "out", "out2"); + } + } + + @Test + public void healthIndicatorDownMultipleKStreamsTest() throws Exception { + try (ConfigurableApplicationContext context = multipleStream()) { + receive(context, + Lists.newArrayList(new ProducerRecord<>("in", "{\"id\":\"123\"}"), + new ProducerRecord<>("in2", "{\"id\":\"124\"}")), + Status.DOWN, "out", "out2"); + } + } + + private static Status getStatusKStream(Map details) { + Health health = (Health) details.get("kstream"); + return health != null ? health.getStatus() : Status.DOWN; + } + + private static boolean waitFor(Map details) { + Health health = (Health) details.get("kstream"); + if (health.getStatus() == Status.UP) { + Map moreDetails = health.getDetails(); + Health kStreamHealth = (Health) moreDetails + .get("kafkaStreamsBinderHealthIndicator"); + String status = (String) kStreamHealth.getDetails().get("threadState"); + return status != null + && (status.equalsIgnoreCase(KafkaStreams.State.REBALANCING.name()) + || status.equalsIgnoreCase("PARTITIONS_REVOKED") + || status.equalsIgnoreCase("PARTITIONS_ASSIGNED") + || status.equalsIgnoreCase( + KafkaStreams.State.PENDING_SHUTDOWN.name())); + } + return false; + } + + private void receive(ConfigurableApplicationContext context, + List> records, Status expected, + String... topics) throws Exception { + Map consumerProps = KafkaTestUtils.consumerProps("group-id0", + "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>( + consumerProps); + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( + senderProps); + try (Consumer consumer = cf.createConsumer()) { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + CountDownLatch latch = new CountDownLatch(records.size()); + for (ProducerRecord record : records) { + ListenableFuture> future = template + .send(record); + future.addCallback( + new ListenableFutureCallback>() { + @Override + public void onFailure(Throwable ex) { + Assert.fail(); + } + + @Override + public void onSuccess(SendResult result) { + latch.countDown(); + } + }); + } + + latch.await(5, TimeUnit.SECONDS); + + embeddedKafka.consumeFromEmbeddedTopics(consumer, topics); + KafkaTestUtils.getRecords(consumer, 1000); + + TimeUnit.SECONDS.sleep(2); + checkHealth(context, expected); + } + finally { + pf.destroy(); + } + } + + private static void checkHealth(ConfigurableApplicationContext context, + Status expected) throws InterruptedException { + HealthIndicator healthIndicator = context.getBean("bindersHealthIndicator", + HealthIndicator.class); + Health health = healthIndicator.health(); + while (waitFor(health.getDetails())) { + TimeUnit.SECONDS.sleep(2); + health = healthIndicator.health(); + } + assertThat(health.getStatus()).isEqualTo(expected); + assertThat(getStatusKStream(health.getDetails())).isEqualTo(expected); + } + + private ConfigurableApplicationContext singleStream() { + SpringApplication app = new SpringApplication(KStreamApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); + return app.run("--server.port=0", "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.input.destination=in", + "--spring.cloud.stream.bindings.output.destination=out", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=" + + "org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=" + + "org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde=" + + "org.apache.kafka.common.serialization.Serdes$IntegerSerde", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=" + + "ApplicationHealthTest-xyz", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + + embeddedKafka.getZookeeperConnectionString()); + } + + private ConfigurableApplicationContext multipleStream() { + System.setProperty("logging.level.org.apache.kafka", "OFF"); + SpringApplication app = new SpringApplication(AnotherKStreamApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); + return app.run("--server.port=0", "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.input.destination=in", + "--spring.cloud.stream.bindings.output.destination=out", + "--spring.cloud.stream.bindings.input2.destination=in2", + "--spring.cloud.stream.bindings.output2.destination=out2", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=" + + "org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=" + + "org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde=" + + "org.apache.kafka.common.serialization.Serdes$IntegerSerde", + "--spring.cloud.stream.kafka.streams.bindings.output2.producer.keySerde=" + + "org.apache.kafka.common.serialization.Serdes$IntegerSerde", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=" + + "ApplicationHealthTest-xyz", + "--spring.cloud.stream.kafka.streams.bindings.input2.consumer.applicationId=" + + "ApplicationHealthTest2-xyz", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + + embeddedKafka.getZookeeperConnectionString()); + } + + @EnableBinding(KafkaStreamsProcessor.class) + @EnableAutoConfiguration + @EnableConfigurationProperties(KafkaStreamsApplicationSupportProperties.class) + public static class KStreamApplication { + + @StreamListener("input") + @SendTo("output") + public KStream process(KStream input) { + return input.filter((key, product) -> { + if (product.getId() != 123) { + throw new IllegalArgumentException(); + } + return true; + }); + } + + } + + @EnableBinding({ KafkaStreamsProcessor.class, KafkaStreamsProcessorX.class }) + @EnableAutoConfiguration + @EnableConfigurationProperties(KafkaStreamsApplicationSupportProperties.class) + public static class AnotherKStreamApplication { + + @StreamListener("input") + @SendTo("output") + public KStream process(KStream input) { + return input.filter((key, product) -> { + if (product.getId() != 123) { + throw new IllegalArgumentException(); + } + return true; + }); + } + + @StreamListener("input2") + @SendTo("output2") + public KStream process2(KStream input) { + return input.filter((key, product) -> { + if (product.getId() != 123) { + throw new IllegalArgumentException(); + } + return true; + }); + } + + } + + public interface KafkaStreamsProcessorX { + + @Input("input2") + KStream input(); + + @Output("output2") + KStream output(); + + } + + static class Product { + + Integer id; + + public Integer getId() { + return id; + } + + public void setId(Integer id) { + this.id = id; + } + + } + +}